feat: add experimental Codex extension and tool (#2320)
This commit is contained in:
Kazuhiro Sera
@@ -0,0 +1,163 @@
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
|
||||
from agents import Agent, Runner, gen_trace_id, trace
|
||||
|
||||
# This tool is still in experimental phase and the details could be changed until being GAed.
|
||||
from agents.extensions.experimental.codex import (
|
||||
CodexToolStreamEvent,
|
||||
CommandExecutionItem,
|
||||
ErrorItem,
|
||||
FileChangeItem,
|
||||
ItemCompletedEvent,
|
||||
ItemStartedEvent,
|
||||
ItemUpdatedEvent,
|
||||
McpToolCallItem,
|
||||
ReasoningItem,
|
||||
ThreadErrorEvent,
|
||||
ThreadOptions,
|
||||
ThreadStartedEvent,
|
||||
TodoListItem,
|
||||
TurnCompletedEvent,
|
||||
TurnFailedEvent,
|
||||
TurnOptions,
|
||||
TurnStartedEvent,
|
||||
WebSearchItem,
|
||||
codex_tool,
|
||||
)
|
||||
|
||||
# This example runs the Codex CLI via the Codex tool wrapper.
|
||||
# You can configure the CLI path with CODEX_PATH or CodexOptions(codex_path_override="...").
|
||||
# codex_tool accepts options as keyword arguments or a plain dict.
|
||||
# For example: codex_tool(sandbox_mode="read-only") or codex_tool({"sandbox_mode": "read-only"}).
|
||||
# The prompt below asks Codex to use the $openai-knowledge skill (Docs MCP) for API lookups.
|
||||
|
||||
|
||||
async def on_codex_stream(payload: CodexToolStreamEvent) -> None:
|
||||
event = payload.event
|
||||
|
||||
if isinstance(event, ThreadStartedEvent):
|
||||
log(f"codex thread started: {event.thread_id}")
|
||||
return
|
||||
if isinstance(event, TurnStartedEvent):
|
||||
log("codex turn started")
|
||||
return
|
||||
if isinstance(event, TurnCompletedEvent):
|
||||
usage = event.usage
|
||||
log(f"codex turn completed, usage: {usage}")
|
||||
return
|
||||
if isinstance(event, TurnFailedEvent):
|
||||
error = event.error.message
|
||||
log(f"codex turn failed: {error}")
|
||||
return
|
||||
if isinstance(event, ThreadErrorEvent):
|
||||
log(f"codex stream error: {event.message}")
|
||||
return
|
||||
|
||||
if not isinstance(event, (ItemStartedEvent, ItemUpdatedEvent, ItemCompletedEvent)):
|
||||
return
|
||||
|
||||
item = event.item
|
||||
|
||||
if isinstance(item, ReasoningItem):
|
||||
text = item.text
|
||||
log(f"codex reasoning ({event.type}): {text}")
|
||||
return
|
||||
if isinstance(item, CommandExecutionItem):
|
||||
command = item.command
|
||||
output = item.aggregated_output
|
||||
output_preview = output[-200:] if isinstance(output, str) else ""
|
||||
status = item.status
|
||||
log(f"codex command {event.type}: {command} | status={status} | output={output_preview}")
|
||||
return
|
||||
if isinstance(item, McpToolCallItem):
|
||||
server = item.server
|
||||
tool = item.tool
|
||||
status = item.status
|
||||
log(f"codex mcp {event.type}: {server}.{tool} | status={status}")
|
||||
return
|
||||
if isinstance(item, FileChangeItem):
|
||||
changes = item.changes
|
||||
status = item.status
|
||||
log(f"codex file change {event.type}: {status} | {changes}")
|
||||
return
|
||||
if isinstance(item, WebSearchItem):
|
||||
log(f"codex web search {event.type}: {item.query}")
|
||||
return
|
||||
if isinstance(item, TodoListItem):
|
||||
items = item.items
|
||||
log(f"codex todo list {event.type}: {len(items)} items")
|
||||
return
|
||||
if isinstance(item, ErrorItem):
|
||||
log(f"codex error {event.type}: {item.message}")
|
||||
|
||||
|
||||
def _timestamp() -> str:
|
||||
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
|
||||
def log(message: str) -> None:
|
||||
timestamp = _timestamp()
|
||||
lines = str(message).splitlines() or [""]
|
||||
for line in lines:
|
||||
print(f"{timestamp} {line}")
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
agent = Agent(
|
||||
name="Codex Agent",
|
||||
instructions=(
|
||||
"Use the codex tool to inspect the workspace and answer the question. "
|
||||
"When skill names, which usually starts with `$`, are mentioned, "
|
||||
"you must rely on the codex tool to use the skill and answer the question.\n\n"
|
||||
"When you send the final answer, you must include the following info at the end:\n\n"
|
||||
"Run `codex resume <thread_id>` to continue the codex session."
|
||||
),
|
||||
tools=[
|
||||
# Run local Codex CLI as a sub process
|
||||
codex_tool(
|
||||
sandbox_mode="workspace-write",
|
||||
default_thread_options=ThreadOptions(
|
||||
# You can pass a Codex instance to customize CLI details
|
||||
# codex=Codex(executable_path="/path/to/codex", base_url="..."),
|
||||
model="gpt-5.2-codex",
|
||||
model_reasoning_effort="low",
|
||||
network_access_enabled=True,
|
||||
web_search_enabled=False,
|
||||
approval_policy="never", # We'll update this example once the HITL is implemented
|
||||
),
|
||||
default_turn_options=TurnOptions(
|
||||
# Abort Codex CLI if no events arrive within this many seconds.
|
||||
idle_timeout_seconds=60,
|
||||
),
|
||||
on_stream=on_codex_stream,
|
||||
)
|
||||
],
|
||||
)
|
||||
trace_id = gen_trace_id()
|
||||
log(f"View trace: https://platform.openai.com/traces/trace?trace_id={trace_id}")
|
||||
|
||||
with trace("Codex tool example", trace_id=trace_id):
|
||||
# Use a skill that requires network access and MCP server settings
|
||||
log("Using $openai-knowledge skill to fetch the latest realtime model name...")
|
||||
result = await Runner.run(
|
||||
agent,
|
||||
"You must use `$openai-knowledge` skill to fetch the latest realtime model name.",
|
||||
)
|
||||
log(result.final_output)
|
||||
# The latest realtime model name, according to the $openai-knowledge skill, is gpt-realtime.
|
||||
|
||||
# Use a skill that runs local command and analyzes the output
|
||||
log(
|
||||
"Using $test-coverage-improver skill to analyze the test coverage of the project and improve it..."
|
||||
)
|
||||
result = await Runner.run(
|
||||
agent,
|
||||
"You must use `$test-coverage-improver` skill to analyze the test coverage of the project and improve it.",
|
||||
)
|
||||
log(result.final_output)
|
||||
# (Aa few suggestions for improving the test coverage will be displayed.)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,6 @@
|
||||
# This package contains experimental extensions to the agents package.
|
||||
# The interface and implementation details could be changed until being GAed.
|
||||
|
||||
__all__ = [
|
||||
"codex",
|
||||
]
|
||||
@@ -0,0 +1,92 @@
|
||||
from .codex import Codex
|
||||
from .codex_options import CodexOptions
|
||||
from .codex_tool import (
|
||||
CodexToolOptions,
|
||||
CodexToolResult,
|
||||
CodexToolStreamEvent,
|
||||
OutputSchemaDescriptor,
|
||||
codex_tool,
|
||||
)
|
||||
from .events import (
|
||||
ItemCompletedEvent,
|
||||
ItemStartedEvent,
|
||||
ItemUpdatedEvent,
|
||||
ThreadError,
|
||||
ThreadErrorEvent,
|
||||
ThreadEvent,
|
||||
ThreadStartedEvent,
|
||||
TurnCompletedEvent,
|
||||
TurnFailedEvent,
|
||||
TurnStartedEvent,
|
||||
Usage,
|
||||
)
|
||||
from .items import (
|
||||
AgentMessageItem,
|
||||
CommandExecutionItem,
|
||||
ErrorItem,
|
||||
FileChangeItem,
|
||||
FileUpdateChange,
|
||||
McpToolCallError,
|
||||
McpToolCallItem,
|
||||
McpToolCallResult,
|
||||
ReasoningItem,
|
||||
ThreadItem,
|
||||
TodoItem,
|
||||
TodoListItem,
|
||||
WebSearchItem,
|
||||
)
|
||||
from .thread import Input, RunResult, RunStreamedResult, Thread, Turn, UserInput
|
||||
from .thread_options import (
|
||||
ApprovalMode,
|
||||
ModelReasoningEffort,
|
||||
SandboxMode,
|
||||
ThreadOptions,
|
||||
WebSearchMode,
|
||||
)
|
||||
from .turn_options import TurnOptions
|
||||
|
||||
__all__ = [
|
||||
"Codex",
|
||||
"CodexOptions",
|
||||
"Thread",
|
||||
"Turn",
|
||||
"RunResult",
|
||||
"RunStreamedResult",
|
||||
"Input",
|
||||
"UserInput",
|
||||
"ThreadOptions",
|
||||
"TurnOptions",
|
||||
"ApprovalMode",
|
||||
"SandboxMode",
|
||||
"ModelReasoningEffort",
|
||||
"WebSearchMode",
|
||||
"ThreadEvent",
|
||||
"ThreadStartedEvent",
|
||||
"TurnStartedEvent",
|
||||
"TurnCompletedEvent",
|
||||
"TurnFailedEvent",
|
||||
"ItemStartedEvent",
|
||||
"ItemUpdatedEvent",
|
||||
"ItemCompletedEvent",
|
||||
"ThreadError",
|
||||
"ThreadErrorEvent",
|
||||
"Usage",
|
||||
"ThreadItem",
|
||||
"AgentMessageItem",
|
||||
"ReasoningItem",
|
||||
"CommandExecutionItem",
|
||||
"FileChangeItem",
|
||||
"FileUpdateChange",
|
||||
"McpToolCallItem",
|
||||
"McpToolCallResult",
|
||||
"McpToolCallError",
|
||||
"WebSearchItem",
|
||||
"TodoItem",
|
||||
"TodoListItem",
|
||||
"ErrorItem",
|
||||
"codex_tool",
|
||||
"CodexToolOptions",
|
||||
"CodexToolResult",
|
||||
"CodexToolStreamEvent",
|
||||
"OutputSchemaDescriptor",
|
||||
]
|
||||
@@ -0,0 +1,89 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, overload
|
||||
|
||||
from agents.exceptions import UserError
|
||||
|
||||
from .codex_options import CodexOptions, coerce_codex_options
|
||||
from .exec import CodexExec
|
||||
from .thread import Thread
|
||||
from .thread_options import ThreadOptions, coerce_thread_options
|
||||
|
||||
|
||||
class _UnsetType:
|
||||
pass
|
||||
|
||||
|
||||
_UNSET = _UnsetType()
|
||||
|
||||
|
||||
class Codex:
|
||||
@overload
|
||||
def __init__(self, options: CodexOptions | Mapping[str, Any] | None = None) -> None: ...
|
||||
|
||||
@overload
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
codex_path_override: str | None = None,
|
||||
base_url: str | None = None,
|
||||
api_key: str | None = None,
|
||||
env: Mapping[str, str] | None = None,
|
||||
) -> None: ...
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
options: CodexOptions | Mapping[str, Any] | None = None,
|
||||
*,
|
||||
codex_path_override: str | None | _UnsetType = _UNSET,
|
||||
base_url: str | None | _UnsetType = _UNSET,
|
||||
api_key: str | None | _UnsetType = _UNSET,
|
||||
env: Mapping[str, str] | None | _UnsetType = _UNSET,
|
||||
) -> None:
|
||||
kw_values = {
|
||||
"codex_path_override": codex_path_override,
|
||||
"base_url": base_url,
|
||||
"api_key": api_key,
|
||||
"env": env,
|
||||
}
|
||||
has_kwargs = any(value is not _UNSET for value in kw_values.values())
|
||||
if options is not None and has_kwargs:
|
||||
raise UserError(
|
||||
"Codex options must be provided as a CodexOptions/mapping or keyword arguments, "
|
||||
"not both."
|
||||
)
|
||||
if has_kwargs:
|
||||
options = {key: value for key, value in kw_values.items() if value is not _UNSET}
|
||||
resolved_options = coerce_codex_options(options) or CodexOptions()
|
||||
self._exec = CodexExec(
|
||||
executable_path=resolved_options.codex_path_override,
|
||||
env=_normalize_env(resolved_options),
|
||||
)
|
||||
self._options = resolved_options
|
||||
|
||||
def start_thread(self, options: ThreadOptions | Mapping[str, Any] | None = None) -> Thread:
|
||||
resolved_options = coerce_thread_options(options) or ThreadOptions()
|
||||
return Thread(
|
||||
exec_client=self._exec,
|
||||
options=self._options,
|
||||
thread_options=resolved_options,
|
||||
)
|
||||
|
||||
def resume_thread(
|
||||
self, thread_id: str, options: ThreadOptions | Mapping[str, Any] | None = None
|
||||
) -> Thread:
|
||||
resolved_options = coerce_thread_options(options) or ThreadOptions()
|
||||
return Thread(
|
||||
exec_client=self._exec,
|
||||
options=self._options,
|
||||
thread_options=resolved_options,
|
||||
thread_id=thread_id,
|
||||
)
|
||||
|
||||
|
||||
def _normalize_env(options: CodexOptions) -> dict[str, str] | None:
|
||||
if options.env is None:
|
||||
return None
|
||||
# Normalize mapping values to strings for subprocess environment.
|
||||
return {str(key): str(value) for key, value in options.env.items()}
|
||||
@@ -0,0 +1,35 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, fields
|
||||
from typing import Any
|
||||
|
||||
from agents.exceptions import UserError
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CodexOptions:
|
||||
# Optional absolute path to the codex CLI binary.
|
||||
codex_path_override: str | None = None
|
||||
# Override OpenAI base URL for the Codex CLI process.
|
||||
base_url: str | None = None
|
||||
# API key passed to the Codex CLI (CODEX_API_KEY).
|
||||
api_key: str | None = None
|
||||
# Environment variables for the Codex CLI process (do not inherit os.environ).
|
||||
env: Mapping[str, str] | None = None
|
||||
|
||||
|
||||
def coerce_codex_options(
|
||||
options: CodexOptions | Mapping[str, Any] | None,
|
||||
) -> CodexOptions | None:
|
||||
if options is None or isinstance(options, CodexOptions):
|
||||
return options
|
||||
if not isinstance(options, Mapping):
|
||||
raise UserError("CodexOptions must be a CodexOptions or a mapping.")
|
||||
|
||||
allowed = {field.name for field in fields(CodexOptions)}
|
||||
unknown = set(options.keys()) - allowed
|
||||
if unknown:
|
||||
raise UserError(f"Unknown CodexOptions field(s): {sorted(unknown)}")
|
||||
|
||||
return CodexOptions(**dict(options))
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,162 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Union, cast
|
||||
|
||||
from typing_extensions import Literal, TypeAlias
|
||||
|
||||
from .items import ThreadItem, coerce_thread_item
|
||||
from .payloads import _DictLike
|
||||
|
||||
# Event payloads emitted by the Codex CLI JSONL stream.
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ThreadStartedEvent(_DictLike):
|
||||
thread_id: str
|
||||
type: Literal["thread.started"] = field(default="thread.started", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TurnStartedEvent(_DictLike):
|
||||
type: Literal["turn.started"] = field(default="turn.started", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Usage(_DictLike):
|
||||
input_tokens: int
|
||||
cached_input_tokens: int
|
||||
output_tokens: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TurnCompletedEvent(_DictLike):
|
||||
usage: Usage | None = None
|
||||
type: Literal["turn.completed"] = field(default="turn.completed", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ThreadError(_DictLike):
|
||||
message: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TurnFailedEvent(_DictLike):
|
||||
error: ThreadError
|
||||
type: Literal["turn.failed"] = field(default="turn.failed", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ItemStartedEvent(_DictLike):
|
||||
item: ThreadItem
|
||||
type: Literal["item.started"] = field(default="item.started", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ItemUpdatedEvent(_DictLike):
|
||||
item: ThreadItem
|
||||
type: Literal["item.updated"] = field(default="item.updated", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ItemCompletedEvent(_DictLike):
|
||||
item: ThreadItem
|
||||
type: Literal["item.completed"] = field(default="item.completed", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ThreadErrorEvent(_DictLike):
|
||||
message: str
|
||||
type: Literal["error"] = field(default="error", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _UnknownThreadEvent(_DictLike):
|
||||
type: str
|
||||
payload: Mapping[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
ThreadEvent: TypeAlias = Union[
|
||||
ThreadStartedEvent,
|
||||
TurnStartedEvent,
|
||||
TurnCompletedEvent,
|
||||
TurnFailedEvent,
|
||||
ItemStartedEvent,
|
||||
ItemUpdatedEvent,
|
||||
ItemCompletedEvent,
|
||||
ThreadErrorEvent,
|
||||
_UnknownThreadEvent,
|
||||
]
|
||||
|
||||
|
||||
def _coerce_thread_error(raw: ThreadError | Mapping[str, Any]) -> ThreadError:
|
||||
if isinstance(raw, ThreadError):
|
||||
return raw
|
||||
if not isinstance(raw, Mapping):
|
||||
raise TypeError("ThreadError must be a mapping.")
|
||||
return ThreadError(message=cast(str, raw.get("message", "")))
|
||||
|
||||
|
||||
def coerce_usage(raw: Usage | Mapping[str, Any]) -> Usage:
|
||||
if isinstance(raw, Usage):
|
||||
return raw
|
||||
if not isinstance(raw, Mapping):
|
||||
raise TypeError("Usage must be a mapping.")
|
||||
return Usage(
|
||||
input_tokens=cast(int, raw["input_tokens"]),
|
||||
cached_input_tokens=cast(int, raw["cached_input_tokens"]),
|
||||
output_tokens=cast(int, raw["output_tokens"]),
|
||||
)
|
||||
|
||||
|
||||
def coerce_thread_event(raw: ThreadEvent | Mapping[str, Any]) -> ThreadEvent:
|
||||
if isinstance(raw, _DictLike):
|
||||
return raw
|
||||
if not isinstance(raw, Mapping):
|
||||
raise TypeError("Thread event payload must be a mapping.")
|
||||
|
||||
event_type = raw.get("type")
|
||||
if event_type == "thread.started":
|
||||
return ThreadStartedEvent(thread_id=cast(str, raw["thread_id"]))
|
||||
if event_type == "turn.started":
|
||||
return TurnStartedEvent()
|
||||
if event_type == "turn.completed":
|
||||
usage_raw = raw.get("usage")
|
||||
usage = coerce_usage(cast(Mapping[str, Any], usage_raw)) if usage_raw is not None else None
|
||||
return TurnCompletedEvent(usage=usage)
|
||||
if event_type == "turn.failed":
|
||||
error_raw = raw.get("error", {})
|
||||
error = _coerce_thread_error(cast(Mapping[str, Any], error_raw))
|
||||
return TurnFailedEvent(error=error)
|
||||
if event_type == "item.started":
|
||||
item_raw = raw.get("item")
|
||||
item = (
|
||||
coerce_thread_item(cast(Union[ThreadItem, Mapping[str, Any]], item_raw))
|
||||
if item_raw is not None
|
||||
else coerce_thread_item({"type": "unknown"})
|
||||
)
|
||||
return ItemStartedEvent(item=item)
|
||||
if event_type == "item.updated":
|
||||
item_raw = raw.get("item")
|
||||
item = (
|
||||
coerce_thread_item(cast(Union[ThreadItem, Mapping[str, Any]], item_raw))
|
||||
if item_raw is not None
|
||||
else coerce_thread_item({"type": "unknown"})
|
||||
)
|
||||
return ItemUpdatedEvent(item=item)
|
||||
if event_type == "item.completed":
|
||||
item_raw = raw.get("item")
|
||||
item = (
|
||||
coerce_thread_item(cast(Union[ThreadItem, Mapping[str, Any]], item_raw))
|
||||
if item_raw is not None
|
||||
else coerce_thread_item({"type": "unknown"})
|
||||
)
|
||||
return ItemCompletedEvent(item=item)
|
||||
if event_type == "error":
|
||||
return ThreadErrorEvent(message=cast(str, raw.get("message", "")))
|
||||
|
||||
return _UnknownThreadEvent(
|
||||
type=cast(str, event_type) if event_type is not None else "unknown",
|
||||
payload=dict(raw),
|
||||
)
|
||||
@@ -0,0 +1,263 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import sys
|
||||
from collections.abc import AsyncGenerator
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from .thread_options import ApprovalMode, ModelReasoningEffort, SandboxMode, WebSearchMode
|
||||
|
||||
_INTERNAL_ORIGINATOR_ENV = "CODEX_INTERNAL_ORIGINATOR_OVERRIDE"
|
||||
_TYPESCRIPT_SDK_ORIGINATOR = "codex_sdk_ts"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CodexExecArgs:
|
||||
input: str
|
||||
base_url: str | None = None
|
||||
api_key: str | None = None
|
||||
thread_id: str | None = None
|
||||
images: list[str] | None = None
|
||||
model: str | None = None
|
||||
sandbox_mode: SandboxMode | None = None
|
||||
working_directory: str | None = None
|
||||
additional_directories: list[str] | None = None
|
||||
skip_git_repo_check: bool | None = None
|
||||
output_schema_file: str | None = None
|
||||
model_reasoning_effort: ModelReasoningEffort | None = None
|
||||
signal: asyncio.Event | None = None
|
||||
idle_timeout_seconds: float | None = None
|
||||
network_access_enabled: bool | None = None
|
||||
web_search_mode: WebSearchMode | None = None
|
||||
web_search_enabled: bool | None = None
|
||||
approval_policy: ApprovalMode | None = None
|
||||
|
||||
|
||||
class CodexExec:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
executable_path: str | None = None,
|
||||
env: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
self._executable_path = executable_path or find_codex_path()
|
||||
self._env_override = env
|
||||
|
||||
async def run(self, args: CodexExecArgs) -> AsyncGenerator[str, None]:
|
||||
# Build the CLI args for `codex exec --experimental-json`.
|
||||
command_args: list[str] = ["exec", "--experimental-json"]
|
||||
|
||||
if args.model:
|
||||
command_args.extend(["--model", args.model])
|
||||
|
||||
if args.sandbox_mode:
|
||||
command_args.extend(["--sandbox", args.sandbox_mode])
|
||||
|
||||
if args.working_directory:
|
||||
command_args.extend(["--cd", args.working_directory])
|
||||
|
||||
if args.additional_directories:
|
||||
for directory in args.additional_directories:
|
||||
command_args.extend(["--add-dir", directory])
|
||||
|
||||
if args.skip_git_repo_check:
|
||||
command_args.append("--skip-git-repo-check")
|
||||
|
||||
if args.output_schema_file:
|
||||
command_args.extend(["--output-schema", args.output_schema_file])
|
||||
|
||||
if args.model_reasoning_effort:
|
||||
command_args.extend(
|
||||
["--config", f'model_reasoning_effort="{args.model_reasoning_effort}"']
|
||||
)
|
||||
|
||||
if args.network_access_enabled is not None:
|
||||
command_args.extend(
|
||||
[
|
||||
"--config",
|
||||
f"sandbox_workspace_write.network_access={str(args.network_access_enabled).lower()}",
|
||||
]
|
||||
)
|
||||
|
||||
if args.web_search_mode:
|
||||
command_args.extend(["--config", f'web_search="{args.web_search_mode}"'])
|
||||
elif args.web_search_enabled is True:
|
||||
command_args.extend(["--config", 'web_search="live"'])
|
||||
elif args.web_search_enabled is False:
|
||||
command_args.extend(["--config", 'web_search="disabled"'])
|
||||
|
||||
if args.approval_policy:
|
||||
command_args.extend(["--config", f'approval_policy="{args.approval_policy}"'])
|
||||
|
||||
if args.thread_id:
|
||||
command_args.extend(["resume", args.thread_id])
|
||||
|
||||
if args.images:
|
||||
for image in args.images:
|
||||
command_args.extend(["--image", image])
|
||||
|
||||
# Codex CLI expects a prompt argument; "-" tells it to read from stdin.
|
||||
command_args.append("-")
|
||||
|
||||
env = self._build_env(args)
|
||||
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
self._executable_path,
|
||||
*command_args,
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
env=env,
|
||||
)
|
||||
|
||||
stderr_chunks: list[bytes] = []
|
||||
|
||||
async def _drain_stderr() -> None:
|
||||
# Preserve stderr for error reporting without blocking stdout reads.
|
||||
if process.stderr is None:
|
||||
return
|
||||
while True:
|
||||
chunk = await process.stderr.read(1024)
|
||||
if not chunk:
|
||||
break
|
||||
stderr_chunks.append(chunk)
|
||||
|
||||
stderr_task = asyncio.create_task(_drain_stderr())
|
||||
|
||||
if process.stdin is None:
|
||||
process.kill()
|
||||
raise RuntimeError("Codex subprocess has no stdin")
|
||||
|
||||
process.stdin.write(args.input.encode("utf-8"))
|
||||
await process.stdin.drain()
|
||||
process.stdin.close()
|
||||
|
||||
if process.stdout is None:
|
||||
process.kill()
|
||||
raise RuntimeError("Codex subprocess has no stdout")
|
||||
stdout = process.stdout
|
||||
|
||||
cancel_task: asyncio.Task[None] | None = None
|
||||
if args.signal is not None:
|
||||
# Mirror AbortSignal semantics by terminating the subprocess.
|
||||
cancel_task = asyncio.create_task(_watch_signal(args.signal, process))
|
||||
|
||||
async def _read_stdout_line() -> bytes:
|
||||
if args.idle_timeout_seconds is None:
|
||||
return await stdout.readline()
|
||||
|
||||
read_task: asyncio.Task[bytes] = asyncio.create_task(stdout.readline())
|
||||
done, _ = await asyncio.wait(
|
||||
{read_task}, timeout=args.idle_timeout_seconds, return_when=asyncio.FIRST_COMPLETED
|
||||
)
|
||||
if read_task in done:
|
||||
return read_task.result()
|
||||
|
||||
if args.signal is not None:
|
||||
args.signal.set()
|
||||
if process.returncode is None:
|
||||
process.terminate()
|
||||
|
||||
read_task.cancel()
|
||||
with contextlib.suppress(asyncio.CancelledError, asyncio.TimeoutError):
|
||||
await asyncio.wait_for(read_task, timeout=1)
|
||||
|
||||
raise RuntimeError(f"Codex stream idle for {args.idle_timeout_seconds} seconds.")
|
||||
|
||||
try:
|
||||
while True:
|
||||
line = await _read_stdout_line()
|
||||
if not line:
|
||||
break
|
||||
yield line.decode("utf-8").rstrip("\n")
|
||||
|
||||
await process.wait()
|
||||
if cancel_task is not None:
|
||||
cancel_task.cancel()
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await cancel_task
|
||||
|
||||
if process.returncode not in (0, None):
|
||||
await stderr_task
|
||||
stderr_text = b"".join(stderr_chunks).decode("utf-8")
|
||||
raise RuntimeError(
|
||||
f"Codex exec exited with code {process.returncode}: {stderr_text}"
|
||||
)
|
||||
finally:
|
||||
if cancel_task is not None and not cancel_task.done():
|
||||
cancel_task.cancel()
|
||||
await stderr_task
|
||||
if process.returncode is None:
|
||||
process.kill()
|
||||
|
||||
def _build_env(self, args: CodexExecArgs) -> dict[str, str]:
|
||||
# Respect env overrides when provided; otherwise copy from os.environ.
|
||||
env: dict[str, str] = {}
|
||||
if self._env_override is not None:
|
||||
env.update(self._env_override)
|
||||
else:
|
||||
env.update({key: value for key, value in os.environ.items() if value is not None})
|
||||
|
||||
# Preserve originator metadata used by the CLI.
|
||||
if _INTERNAL_ORIGINATOR_ENV not in env:
|
||||
env[_INTERNAL_ORIGINATOR_ENV] = _TYPESCRIPT_SDK_ORIGINATOR
|
||||
|
||||
if args.base_url:
|
||||
env["OPENAI_BASE_URL"] = args.base_url
|
||||
if args.api_key:
|
||||
env["CODEX_API_KEY"] = args.api_key
|
||||
|
||||
return env
|
||||
|
||||
|
||||
async def _watch_signal(signal: asyncio.Event, process: asyncio.subprocess.Process) -> None:
|
||||
await signal.wait()
|
||||
if process.returncode is None:
|
||||
process.terminate()
|
||||
|
||||
|
||||
def _platform_target_triple() -> str:
|
||||
# Map the running platform to the vendor layout used in Codex releases.
|
||||
system = sys.platform
|
||||
arch = platform.machine().lower()
|
||||
|
||||
if system.startswith("linux"):
|
||||
if arch in {"x86_64", "amd64"}:
|
||||
return "x86_64-unknown-linux-musl"
|
||||
if arch in {"aarch64", "arm64"}:
|
||||
return "aarch64-unknown-linux-musl"
|
||||
if system == "darwin":
|
||||
if arch in {"x86_64", "amd64"}:
|
||||
return "x86_64-apple-darwin"
|
||||
if arch in {"arm64", "aarch64"}:
|
||||
return "aarch64-apple-darwin"
|
||||
if system in {"win32", "cygwin"}:
|
||||
if arch in {"x86_64", "amd64"}:
|
||||
return "x86_64-pc-windows-msvc"
|
||||
if arch in {"arm64", "aarch64"}:
|
||||
return "aarch64-pc-windows-msvc"
|
||||
|
||||
raise RuntimeError(f"Unsupported platform: {system} ({arch})")
|
||||
|
||||
|
||||
def find_codex_path() -> str:
|
||||
# Resolution order: CODEX_PATH env, PATH lookup, bundled vendor binary.
|
||||
path_override = os.environ.get("CODEX_PATH")
|
||||
if path_override:
|
||||
return path_override
|
||||
|
||||
which_path = shutil.which("codex")
|
||||
if which_path:
|
||||
return which_path
|
||||
|
||||
target_triple = _platform_target_triple()
|
||||
vendor_root = Path(__file__).resolve().parent.parent.parent / "vendor"
|
||||
arch_root = vendor_root / target_triple
|
||||
binary_name = "codex.exe" if sys.platform.startswith("win") else "codex"
|
||||
binary_path = arch_root / "codex" / binary_name
|
||||
return str(binary_path)
|
||||
@@ -0,0 +1,245 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from typing import TYPE_CHECKING, Any, Optional, Union, cast
|
||||
|
||||
from typing_extensions import Literal, TypeAlias, TypeGuard
|
||||
|
||||
from .payloads import _DictLike
|
||||
|
||||
# Item payloads are emitted inside item.* events from the Codex CLI JSONL stream.
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from mcp.types import ContentBlock as McpContentBlock
|
||||
else:
|
||||
McpContentBlock = Any # type: ignore[assignment]
|
||||
|
||||
CommandExecutionStatus = Literal["in_progress", "completed", "failed"]
|
||||
PatchChangeKind = Literal["add", "delete", "update"]
|
||||
PatchApplyStatus = Literal["completed", "failed"]
|
||||
McpToolCallStatus = Literal["in_progress", "completed", "failed"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CommandExecutionItem(_DictLike):
|
||||
id: str
|
||||
command: str
|
||||
status: CommandExecutionStatus
|
||||
aggregated_output: str = ""
|
||||
exit_code: int | None = None
|
||||
type: Literal["command_execution"] = field(default="command_execution", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FileUpdateChange(_DictLike):
|
||||
path: str
|
||||
kind: PatchChangeKind
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FileChangeItem(_DictLike):
|
||||
id: str
|
||||
changes: list[FileUpdateChange]
|
||||
status: PatchApplyStatus
|
||||
type: Literal["file_change"] = field(default="file_change", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class McpToolCallResult(_DictLike):
|
||||
content: list[McpContentBlock]
|
||||
structured_content: Any
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class McpToolCallError(_DictLike):
|
||||
message: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class McpToolCallItem(_DictLike):
|
||||
id: str
|
||||
server: str
|
||||
tool: str
|
||||
arguments: Any
|
||||
status: McpToolCallStatus
|
||||
result: McpToolCallResult | None = None
|
||||
error: McpToolCallError | None = None
|
||||
type: Literal["mcp_tool_call"] = field(default="mcp_tool_call", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AgentMessageItem(_DictLike):
|
||||
id: str
|
||||
text: str
|
||||
type: Literal["agent_message"] = field(default="agent_message", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReasoningItem(_DictLike):
|
||||
id: str
|
||||
text: str
|
||||
type: Literal["reasoning"] = field(default="reasoning", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class WebSearchItem(_DictLike):
|
||||
id: str
|
||||
query: str
|
||||
type: Literal["web_search"] = field(default="web_search", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ErrorItem(_DictLike):
|
||||
id: str
|
||||
message: str
|
||||
type: Literal["error"] = field(default="error", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TodoItem(_DictLike):
|
||||
text: str
|
||||
completed: bool
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TodoListItem(_DictLike):
|
||||
id: str
|
||||
items: list[TodoItem]
|
||||
type: Literal["todo_list"] = field(default="todo_list", init=False)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _UnknownThreadItem(_DictLike):
|
||||
type: str
|
||||
payload: Mapping[str, Any] = field(default_factory=dict)
|
||||
id: str | None = None
|
||||
|
||||
|
||||
ThreadItem: TypeAlias = Union[
|
||||
AgentMessageItem,
|
||||
ReasoningItem,
|
||||
CommandExecutionItem,
|
||||
FileChangeItem,
|
||||
McpToolCallItem,
|
||||
WebSearchItem,
|
||||
TodoListItem,
|
||||
ErrorItem,
|
||||
_UnknownThreadItem,
|
||||
]
|
||||
|
||||
|
||||
def is_agent_message_item(item: ThreadItem) -> TypeGuard[AgentMessageItem]:
|
||||
return isinstance(item, AgentMessageItem)
|
||||
|
||||
|
||||
def _coerce_file_update_change(
|
||||
raw: FileUpdateChange | Mapping[str, Any],
|
||||
) -> FileUpdateChange:
|
||||
if isinstance(raw, FileUpdateChange):
|
||||
return raw
|
||||
if not isinstance(raw, Mapping):
|
||||
raise TypeError("FileUpdateChange must be a mapping.")
|
||||
return FileUpdateChange(
|
||||
path=cast(str, raw["path"]),
|
||||
kind=cast(PatchChangeKind, raw["kind"]),
|
||||
)
|
||||
|
||||
|
||||
def _coerce_mcp_tool_call_result(
|
||||
raw: McpToolCallResult | Mapping[str, Any],
|
||||
) -> McpToolCallResult:
|
||||
if isinstance(raw, McpToolCallResult):
|
||||
return raw
|
||||
if not isinstance(raw, Mapping):
|
||||
raise TypeError("McpToolCallResult must be a mapping.")
|
||||
content = cast(list[McpContentBlock], raw.get("content", []))
|
||||
return McpToolCallResult(
|
||||
content=content,
|
||||
structured_content=raw.get("structured_content"),
|
||||
)
|
||||
|
||||
|
||||
def _coerce_mcp_tool_call_error(
|
||||
raw: McpToolCallError | Mapping[str, Any],
|
||||
) -> McpToolCallError:
|
||||
if isinstance(raw, McpToolCallError):
|
||||
return raw
|
||||
if not isinstance(raw, Mapping):
|
||||
raise TypeError("McpToolCallError must be a mapping.")
|
||||
return McpToolCallError(message=cast(str, raw.get("message", "")))
|
||||
|
||||
|
||||
def coerce_thread_item(raw: ThreadItem | Mapping[str, Any]) -> ThreadItem:
|
||||
if isinstance(raw, _DictLike):
|
||||
return raw
|
||||
if not isinstance(raw, Mapping):
|
||||
raise TypeError("Thread item payload must be a mapping.")
|
||||
|
||||
item_type = raw.get("type")
|
||||
if item_type == "command_execution":
|
||||
return CommandExecutionItem(
|
||||
id=cast(str, raw["id"]),
|
||||
command=cast(str, raw["command"]),
|
||||
aggregated_output=cast(str, raw.get("aggregated_output", "")),
|
||||
status=cast(CommandExecutionStatus, raw["status"]),
|
||||
exit_code=cast(Optional[int], raw.get("exit_code")),
|
||||
)
|
||||
if item_type == "file_change":
|
||||
changes = [_coerce_file_update_change(change) for change in raw.get("changes", [])]
|
||||
return FileChangeItem(
|
||||
id=cast(str, raw["id"]),
|
||||
changes=changes,
|
||||
status=cast(PatchApplyStatus, raw["status"]),
|
||||
)
|
||||
if item_type == "mcp_tool_call":
|
||||
result_raw = raw.get("result")
|
||||
error_raw = raw.get("error")
|
||||
result = None
|
||||
error = None
|
||||
if result_raw is not None:
|
||||
result = _coerce_mcp_tool_call_result(cast(Mapping[str, Any], result_raw))
|
||||
if error_raw is not None:
|
||||
error = _coerce_mcp_tool_call_error(cast(Mapping[str, Any], error_raw))
|
||||
return McpToolCallItem(
|
||||
id=cast(str, raw["id"]),
|
||||
server=cast(str, raw["server"]),
|
||||
tool=cast(str, raw["tool"]),
|
||||
arguments=raw.get("arguments"),
|
||||
status=cast(McpToolCallStatus, raw["status"]),
|
||||
result=result,
|
||||
error=error,
|
||||
)
|
||||
if item_type == "agent_message":
|
||||
return AgentMessageItem(
|
||||
id=cast(str, raw["id"]),
|
||||
text=cast(str, raw.get("text", "")),
|
||||
)
|
||||
if item_type == "reasoning":
|
||||
return ReasoningItem(
|
||||
id=cast(str, raw["id"]),
|
||||
text=cast(str, raw.get("text", "")),
|
||||
)
|
||||
if item_type == "web_search":
|
||||
return WebSearchItem(
|
||||
id=cast(str, raw["id"]),
|
||||
query=cast(str, raw.get("query", "")),
|
||||
)
|
||||
if item_type == "todo_list":
|
||||
items_raw = raw.get("items", [])
|
||||
items = [
|
||||
TodoItem(text=cast(str, item.get("text", "")), completed=bool(item.get("completed")))
|
||||
for item in cast(list[Mapping[str, Any]], items_raw)
|
||||
]
|
||||
return TodoListItem(id=cast(str, raw["id"]), items=items)
|
||||
if item_type == "error":
|
||||
return ErrorItem(
|
||||
id=cast(str, raw.get("id", "")),
|
||||
message=cast(str, raw.get("message", "")),
|
||||
)
|
||||
|
||||
return _UnknownThreadItem(
|
||||
type=cast(str, item_type) if item_type is not None else "unknown",
|
||||
payload=dict(raw),
|
||||
id=cast(Optional[str], raw.get("id")),
|
||||
)
|
||||
@@ -0,0 +1,50 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable
|
||||
|
||||
from agents.exceptions import UserError
|
||||
|
||||
|
||||
@dataclass
|
||||
class OutputSchemaFile:
|
||||
# Holds the on-disk schema path and cleanup callback.
|
||||
schema_path: str | None
|
||||
cleanup: Callable[[], None]
|
||||
|
||||
|
||||
def _is_plain_json_object(schema: Any) -> bool:
|
||||
return isinstance(schema, dict)
|
||||
|
||||
|
||||
def create_output_schema_file(schema: dict[str, Any] | None) -> OutputSchemaFile:
|
||||
"""Materialize a JSON schema into a temp file for the Codex CLI."""
|
||||
if schema is None:
|
||||
# No schema means there is no temp file to manage.
|
||||
return OutputSchemaFile(schema_path=None, cleanup=lambda: None)
|
||||
|
||||
if not _is_plain_json_object(schema):
|
||||
raise UserError("output_schema must be a plain JSON object")
|
||||
|
||||
# The Codex CLI expects a schema file path, so write to a temp directory.
|
||||
schema_dir = tempfile.mkdtemp(prefix="codex-output-schema-")
|
||||
schema_path = os.path.join(schema_dir, "schema.json")
|
||||
|
||||
def cleanup() -> None:
|
||||
# Best-effort cleanup since this runs in finally blocks.
|
||||
try:
|
||||
shutil.rmtree(schema_dir, ignore_errors=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
with open(schema_path, "w", encoding="utf-8") as handle:
|
||||
json.dump(schema, handle)
|
||||
return OutputSchemaFile(schema_path=schema_path, cleanup=cleanup)
|
||||
except Exception:
|
||||
cleanup()
|
||||
raise
|
||||
@@ -0,0 +1,31 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
from collections.abc import Iterable
|
||||
from typing import Any, cast
|
||||
|
||||
|
||||
class _DictLike:
|
||||
def __getitem__(self, key: str) -> Any:
|
||||
if key in self._field_names():
|
||||
return getattr(self, key)
|
||||
raise KeyError(key)
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
if key in self._field_names():
|
||||
return getattr(self, key)
|
||||
return default
|
||||
|
||||
def __contains__(self, key: object) -> bool:
|
||||
if not isinstance(key, str):
|
||||
return False
|
||||
return key in self._field_names()
|
||||
|
||||
def keys(self) -> Iterable[str]:
|
||||
return iter(self._field_names())
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return dataclasses.asdict(cast(Any, self))
|
||||
|
||||
def _field_names(self) -> list[str]:
|
||||
return [field.name for field in dataclasses.fields(cast(Any, self))]
|
||||
@@ -0,0 +1,214 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
from collections.abc import AsyncGenerator
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Union, cast
|
||||
|
||||
from typing_extensions import Literal, TypeAlias, TypedDict
|
||||
|
||||
from .codex_options import CodexOptions
|
||||
from .events import (
|
||||
ItemCompletedEvent,
|
||||
ThreadError,
|
||||
ThreadErrorEvent,
|
||||
ThreadEvent,
|
||||
ThreadStartedEvent,
|
||||
TurnCompletedEvent,
|
||||
TurnFailedEvent,
|
||||
Usage,
|
||||
coerce_thread_event,
|
||||
)
|
||||
from .exec import CodexExec, CodexExecArgs
|
||||
from .items import ThreadItem, is_agent_message_item
|
||||
from .output_schema_file import create_output_schema_file
|
||||
from .thread_options import ThreadOptions
|
||||
from .turn_options import TurnOptions
|
||||
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def _aclosing(
|
||||
generator: AsyncGenerator[str, None],
|
||||
) -> AsyncGenerator[AsyncGenerator[str, None], None]:
|
||||
try:
|
||||
yield generator
|
||||
finally:
|
||||
await generator.aclose()
|
||||
|
||||
|
||||
class TextInput(TypedDict):
|
||||
type: Literal["text"]
|
||||
text: str
|
||||
|
||||
|
||||
class LocalImageInput(TypedDict):
|
||||
type: Literal["local_image"]
|
||||
path: str
|
||||
|
||||
|
||||
UserInput: TypeAlias = Union[TextInput, LocalImageInput]
|
||||
Input: TypeAlias = Union[str, list[UserInput]]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Turn:
|
||||
items: list[ThreadItem]
|
||||
final_response: str
|
||||
usage: Usage | None
|
||||
|
||||
|
||||
RunResult = Turn
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class StreamedTurn:
|
||||
events: AsyncGenerator[ThreadEvent, None]
|
||||
|
||||
|
||||
RunStreamedResult = StreamedTurn
|
||||
|
||||
|
||||
class Thread:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
exec_client: CodexExec,
|
||||
options: CodexOptions,
|
||||
thread_options: ThreadOptions,
|
||||
thread_id: str | None = None,
|
||||
) -> None:
|
||||
self._exec = exec_client
|
||||
self._options = options
|
||||
self._id = thread_id
|
||||
self._thread_options = thread_options
|
||||
|
||||
@property
|
||||
def id(self) -> str | None:
|
||||
return self._id
|
||||
|
||||
async def run_streamed(
|
||||
self, input: Input, turn_options: TurnOptions | None = None
|
||||
) -> StreamedTurn:
|
||||
options = turn_options or TurnOptions()
|
||||
return StreamedTurn(events=self._run_streamed_internal(input, options))
|
||||
|
||||
async def _run_streamed_internal(
|
||||
self, input: Input, turn_options: TurnOptions
|
||||
) -> AsyncGenerator[ThreadEvent, None]:
|
||||
# The Codex CLI expects an output schema file path for structured output.
|
||||
output_schema_file = create_output_schema_file(turn_options.output_schema)
|
||||
options = self._thread_options
|
||||
prompt, images = _normalize_input(input)
|
||||
idle_timeout = turn_options.idle_timeout_seconds
|
||||
signal = turn_options.signal
|
||||
if idle_timeout is not None and signal is None:
|
||||
signal = asyncio.Event()
|
||||
generator = self._exec.run(
|
||||
CodexExecArgs(
|
||||
input=prompt,
|
||||
base_url=self._options.base_url,
|
||||
api_key=self._options.api_key,
|
||||
thread_id=self._id,
|
||||
images=images,
|
||||
model=options.model,
|
||||
sandbox_mode=options.sandbox_mode,
|
||||
working_directory=options.working_directory,
|
||||
skip_git_repo_check=options.skip_git_repo_check,
|
||||
output_schema_file=output_schema_file.schema_path,
|
||||
model_reasoning_effort=options.model_reasoning_effort,
|
||||
signal=signal,
|
||||
idle_timeout_seconds=idle_timeout,
|
||||
network_access_enabled=options.network_access_enabled,
|
||||
web_search_mode=options.web_search_mode,
|
||||
web_search_enabled=options.web_search_enabled,
|
||||
approval_policy=options.approval_policy,
|
||||
additional_directories=list(options.additional_directories)
|
||||
if options.additional_directories
|
||||
else None,
|
||||
)
|
||||
)
|
||||
|
||||
try:
|
||||
async with _aclosing(generator) as stream:
|
||||
while True:
|
||||
try:
|
||||
if idle_timeout is None or isinstance(self._exec, CodexExec):
|
||||
item = await stream.__anext__()
|
||||
else:
|
||||
item = await asyncio.wait_for(
|
||||
stream.__anext__(),
|
||||
timeout=idle_timeout,
|
||||
)
|
||||
except StopAsyncIteration:
|
||||
break
|
||||
except asyncio.TimeoutError as exc:
|
||||
if signal is not None:
|
||||
signal.set()
|
||||
raise RuntimeError(
|
||||
f"Codex stream idle for {idle_timeout} seconds."
|
||||
) from exc
|
||||
try:
|
||||
parsed = _parse_event(item)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
raise RuntimeError(f"Failed to parse event: {item}") from exc
|
||||
if isinstance(parsed, ThreadStartedEvent):
|
||||
# Capture the thread id so callers can resume later.
|
||||
self._id = parsed.thread_id
|
||||
yield parsed
|
||||
finally:
|
||||
output_schema_file.cleanup()
|
||||
|
||||
async def run(self, input: Input, turn_options: TurnOptions | None = None) -> Turn:
|
||||
# Aggregate events into a single Turn result (matching the TS SDK behavior).
|
||||
options = turn_options or TurnOptions()
|
||||
generator = self._run_streamed_internal(input, options)
|
||||
items: list[ThreadItem] = []
|
||||
final_response = ""
|
||||
usage: Usage | None = None
|
||||
turn_failure: ThreadError | None = None
|
||||
|
||||
async for event in generator:
|
||||
if isinstance(event, ItemCompletedEvent):
|
||||
item = event.item
|
||||
if is_agent_message_item(item):
|
||||
final_response = item.text
|
||||
items.append(item)
|
||||
elif isinstance(event, TurnCompletedEvent):
|
||||
usage = event.usage
|
||||
elif isinstance(event, TurnFailedEvent):
|
||||
turn_failure = event.error
|
||||
break
|
||||
elif isinstance(event, ThreadErrorEvent):
|
||||
raise RuntimeError(f"Codex stream error: {event.message}")
|
||||
|
||||
if turn_failure:
|
||||
raise RuntimeError(turn_failure.message)
|
||||
|
||||
return Turn(items=items, final_response=final_response, usage=usage)
|
||||
|
||||
|
||||
def _normalize_input(input: Input) -> tuple[str, list[str]]:
|
||||
# Merge text items into a single prompt and collect image paths.
|
||||
if isinstance(input, str):
|
||||
return input, []
|
||||
|
||||
prompt_parts: list[str] = []
|
||||
images: list[str] = []
|
||||
for item in input:
|
||||
if item["type"] == "text":
|
||||
text = item.get("text", "")
|
||||
prompt_parts.append(text)
|
||||
elif item["type"] == "local_image":
|
||||
path = item.get("path", "")
|
||||
if path:
|
||||
images.append(path)
|
||||
|
||||
return "\n\n".join(prompt_parts), images
|
||||
|
||||
|
||||
def _parse_event(raw: str) -> ThreadEvent:
|
||||
import json
|
||||
|
||||
parsed = json.loads(raw)
|
||||
return coerce_thread_event(cast(dict[str, Any], parsed))
|
||||
@@ -0,0 +1,54 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass, fields
|
||||
from typing import Any
|
||||
|
||||
from typing_extensions import Literal
|
||||
|
||||
from agents.exceptions import UserError
|
||||
|
||||
ApprovalMode = Literal["never", "on-request", "on-failure", "untrusted"]
|
||||
SandboxMode = Literal["read-only", "workspace-write", "danger-full-access"]
|
||||
ModelReasoningEffort = Literal["minimal", "low", "medium", "high", "xhigh"]
|
||||
WebSearchMode = Literal["disabled", "cached", "live"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ThreadOptions:
|
||||
# Model identifier passed to the Codex CLI (--model).
|
||||
model: str | None = None
|
||||
# Sandbox permissions for filesystem/network access.
|
||||
sandbox_mode: SandboxMode | None = None
|
||||
# Working directory for the Codex CLI process.
|
||||
working_directory: str | None = None
|
||||
# Allow running outside a Git repository.
|
||||
skip_git_repo_check: bool | None = None
|
||||
# Configure model reasoning effort.
|
||||
model_reasoning_effort: ModelReasoningEffort | None = None
|
||||
# Toggle network access in sandboxed workspace writes.
|
||||
network_access_enabled: bool | None = None
|
||||
# Configure web search mode via codex config.
|
||||
web_search_mode: WebSearchMode | None = None
|
||||
# Legacy toggle for web search behavior.
|
||||
web_search_enabled: bool | None = None
|
||||
# Approval policy for tool invocations within Codex.
|
||||
approval_policy: ApprovalMode | None = None
|
||||
# Additional filesystem roots available to Codex.
|
||||
additional_directories: Sequence[str] | None = None
|
||||
|
||||
|
||||
def coerce_thread_options(
|
||||
options: ThreadOptions | Mapping[str, Any] | None,
|
||||
) -> ThreadOptions | None:
|
||||
if options is None or isinstance(options, ThreadOptions):
|
||||
return options
|
||||
if not isinstance(options, Mapping):
|
||||
raise UserError("ThreadOptions must be a ThreadOptions or a mapping.")
|
||||
|
||||
allowed = {field.name for field in fields(ThreadOptions)}
|
||||
unknown = set(options.keys()) - allowed
|
||||
if unknown:
|
||||
raise UserError(f"Unknown ThreadOptions field(s): {sorted(unknown)}")
|
||||
|
||||
return ThreadOptions(**dict(options))
|
||||
@@ -0,0 +1,36 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, fields
|
||||
from typing import Any
|
||||
|
||||
from agents.exceptions import UserError
|
||||
|
||||
AbortSignal = asyncio.Event
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TurnOptions:
|
||||
# JSON schema used by Codex for structured output.
|
||||
output_schema: dict[str, Any] | None = None
|
||||
# Cancellation signal for the Codex CLI subprocess.
|
||||
signal: AbortSignal | None = None
|
||||
# Abort the Codex CLI if no events arrive within this many seconds.
|
||||
idle_timeout_seconds: float | None = None
|
||||
|
||||
|
||||
def coerce_turn_options(
|
||||
options: TurnOptions | Mapping[str, Any] | None,
|
||||
) -> TurnOptions | None:
|
||||
if options is None or isinstance(options, TurnOptions):
|
||||
return options
|
||||
if not isinstance(options, Mapping):
|
||||
raise UserError("TurnOptions must be a TurnOptions or a mapping.")
|
||||
|
||||
allowed = {field.name for field in fields(TurnOptions)}
|
||||
unknown = set(options.keys()) - allowed
|
||||
if unknown:
|
||||
raise UserError(f"Unknown TurnOptions field(s): {sorted(unknown)}")
|
||||
|
||||
return TurnOptions(**dict(options))
|
||||
@@ -0,0 +1,644 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import importlib
|
||||
import inspect
|
||||
import json
|
||||
import os
|
||||
from dataclasses import fields
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
|
||||
from agents.exceptions import UserError
|
||||
from agents.extensions.experimental.codex import Usage
|
||||
from agents.extensions.experimental.codex.codex import Codex, _normalize_env
|
||||
from agents.extensions.experimental.codex.codex_options import CodexOptions, coerce_codex_options
|
||||
from agents.extensions.experimental.codex.exec import CodexExec
|
||||
from agents.extensions.experimental.codex.output_schema_file import (
|
||||
OutputSchemaFile,
|
||||
create_output_schema_file,
|
||||
)
|
||||
from agents.extensions.experimental.codex.thread import Thread, _normalize_input
|
||||
from agents.extensions.experimental.codex.thread_options import ThreadOptions, coerce_thread_options
|
||||
from agents.extensions.experimental.codex.turn_options import TurnOptions
|
||||
|
||||
exec_module = importlib.import_module("agents.extensions.experimental.codex.exec")
|
||||
thread_module = importlib.import_module("agents.extensions.experimental.codex.thread")
|
||||
output_schema_module = importlib.import_module(
|
||||
"agents.extensions.experimental.codex.output_schema_file"
|
||||
)
|
||||
|
||||
|
||||
class FakeStdin:
|
||||
def __init__(self) -> None:
|
||||
self.buffer = b""
|
||||
self.closed = False
|
||||
|
||||
def write(self, data: bytes) -> None:
|
||||
self.buffer += data
|
||||
|
||||
async def drain(self) -> None:
|
||||
return None
|
||||
|
||||
def close(self) -> None:
|
||||
self.closed = True
|
||||
|
||||
|
||||
class FakeStdout:
|
||||
def __init__(self, lines: list[str]) -> None:
|
||||
self._lines = [line.encode("utf-8") for line in lines]
|
||||
|
||||
async def readline(self) -> bytes:
|
||||
if not self._lines:
|
||||
return b""
|
||||
return self._lines.pop(0)
|
||||
|
||||
|
||||
class FakeStderr:
|
||||
def __init__(self, chunks: list[bytes]) -> None:
|
||||
self._chunks = list(chunks)
|
||||
|
||||
async def read(self, _size: int) -> bytes:
|
||||
if not self._chunks:
|
||||
return b""
|
||||
return self._chunks.pop(0)
|
||||
|
||||
|
||||
class FakeProcess:
|
||||
def __init__(
|
||||
self,
|
||||
stdout_lines: list[str],
|
||||
stderr_chunks: list[bytes] | None = None,
|
||||
*,
|
||||
returncode: int | None = 0,
|
||||
stdin_present: bool = True,
|
||||
stdout_present: bool = True,
|
||||
stderr_present: bool = True,
|
||||
) -> None:
|
||||
self.stdin = FakeStdin() if stdin_present else None
|
||||
self.stdout = FakeStdout(stdout_lines) if stdout_present else None
|
||||
self.stderr = FakeStderr(stderr_chunks or []) if stderr_present else None
|
||||
self.returncode = returncode
|
||||
self.killed = False
|
||||
self.terminated = False
|
||||
|
||||
async def wait(self) -> None:
|
||||
if self.returncode is None:
|
||||
self.returncode = 0
|
||||
|
||||
def kill(self) -> None:
|
||||
self.killed = True
|
||||
|
||||
def terminate(self) -> None:
|
||||
self.terminated = True
|
||||
|
||||
|
||||
class FakeExec:
|
||||
def __init__(self, events: list[Any], delay: float = 0.0) -> None:
|
||||
self.events = events
|
||||
self.delay = delay
|
||||
self.last_args: Any = None
|
||||
|
||||
async def run(self, args: Any):
|
||||
self.last_args = args
|
||||
for event in self.events:
|
||||
if self.delay:
|
||||
await asyncio.sleep(self.delay)
|
||||
payload = event if isinstance(event, str) else json.dumps(event)
|
||||
yield payload
|
||||
|
||||
|
||||
def test_output_schema_file_none_schema() -> None:
|
||||
result = create_output_schema_file(None)
|
||||
assert result.schema_path is None
|
||||
result.cleanup()
|
||||
|
||||
|
||||
def test_output_schema_file_rejects_non_object() -> None:
|
||||
with pytest.raises(UserError, match="output_schema must be a plain JSON object"):
|
||||
create_output_schema_file(cast(Any, ["not", "an", "object"]))
|
||||
|
||||
|
||||
def test_output_schema_file_creates_and_cleans() -> None:
|
||||
schema = {"type": "object", "properties": {"foo": {"type": "string"}}}
|
||||
result = create_output_schema_file(schema)
|
||||
assert result.schema_path is not None
|
||||
with open(result.schema_path, encoding="utf-8") as handle:
|
||||
assert json.load(handle) == schema
|
||||
result.cleanup()
|
||||
assert not os.path.exists(result.schema_path)
|
||||
|
||||
|
||||
def test_output_schema_file_cleanup_swallows_rmtree_errors(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
schema = {"type": "object"}
|
||||
called = False
|
||||
|
||||
def bad_rmtree(_path: str, ignore_errors: bool = True) -> None:
|
||||
nonlocal called
|
||||
called = True
|
||||
raise OSError("boom")
|
||||
|
||||
monkeypatch.setattr(output_schema_module.shutil, "rmtree", bad_rmtree)
|
||||
|
||||
result = create_output_schema_file(schema)
|
||||
result.cleanup()
|
||||
|
||||
assert called is True
|
||||
|
||||
|
||||
def test_output_schema_file_cleanup_on_write_error(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
schema = {"type": "object"}
|
||||
cleanup_called = False
|
||||
|
||||
def bad_dump(*_args: Any, **_kwargs: Any) -> None:
|
||||
raise RuntimeError("boom")
|
||||
|
||||
def fake_rmtree(_path: str, ignore_errors: bool = True) -> None:
|
||||
nonlocal cleanup_called
|
||||
cleanup_called = True
|
||||
|
||||
monkeypatch.setattr(output_schema_module.json, "dump", bad_dump)
|
||||
monkeypatch.setattr(output_schema_module.shutil, "rmtree", fake_rmtree)
|
||||
|
||||
with pytest.raises(RuntimeError, match="boom"):
|
||||
create_output_schema_file(schema)
|
||||
|
||||
assert cleanup_called is True
|
||||
|
||||
|
||||
def test_normalize_input_merges_text_and_images() -> None:
|
||||
prompt, images = _normalize_input(
|
||||
[
|
||||
{"type": "text", "text": "first"},
|
||||
{"type": "local_image", "path": "/tmp/a.png"},
|
||||
{"type": "text", "text": "second"},
|
||||
{"type": "local_image", "path": ""},
|
||||
]
|
||||
)
|
||||
assert prompt == "first\n\nsecond"
|
||||
assert images == ["/tmp/a.png"]
|
||||
|
||||
|
||||
def test_normalize_env_stringifies_values() -> None:
|
||||
env = _normalize_env(CodexOptions(env=cast(dict[str, str], {"FOO": 1, 2: "bar"})))
|
||||
assert env == {"FOO": "1", "2": "bar"}
|
||||
|
||||
|
||||
def test_coerce_codex_options_rejects_unknown_fields() -> None:
|
||||
with pytest.raises(UserError, match="Unknown CodexOptions field"):
|
||||
coerce_codex_options({"unknown": "value"})
|
||||
|
||||
|
||||
def test_coerce_thread_options_rejects_unknown_fields() -> None:
|
||||
with pytest.raises(UserError, match="Unknown ThreadOptions field"):
|
||||
coerce_thread_options({"unknown": "value"})
|
||||
|
||||
|
||||
def test_codex_start_and_resume_thread() -> None:
|
||||
codex = Codex(CodexOptions(codex_path_override="/bin/codex"))
|
||||
thread = codex.start_thread({"model": "gpt"})
|
||||
assert thread.id is None
|
||||
resumed = codex.resume_thread("thread-1", {"model": "gpt"})
|
||||
assert resumed.id == "thread-1"
|
||||
|
||||
|
||||
def test_codex_init_accepts_mapping_options() -> None:
|
||||
codex = Codex({"codex_path_override": "/bin/codex"})
|
||||
assert codex._exec._executable_path == "/bin/codex"
|
||||
|
||||
|
||||
def test_codex_init_accepts_kwargs() -> None:
|
||||
codex = Codex(codex_path_override="/bin/codex", base_url="https://example.com")
|
||||
assert codex._exec._executable_path == "/bin/codex"
|
||||
assert codex._options.base_url == "https://example.com"
|
||||
|
||||
|
||||
def test_codex_init_rejects_options_and_kwargs() -> None:
|
||||
with pytest.raises(UserError, match="Codex options must be provided"):
|
||||
Codex( # type: ignore[call-overload]
|
||||
cast(Any, CodexOptions()), codex_path_override="/bin/codex"
|
||||
)
|
||||
|
||||
|
||||
def test_codex_init_kw_matches_codex_options() -> None:
|
||||
signature = inspect.signature(Codex.__init__)
|
||||
kw_only = [
|
||||
param.name
|
||||
for param in signature.parameters.values()
|
||||
if param.kind == inspect.Parameter.KEYWORD_ONLY
|
||||
]
|
||||
option_fields = [field.name for field in fields(CodexOptions)]
|
||||
assert kw_only == option_fields
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_codex_exec_run_builds_command_args_and_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
captured: dict[str, Any] = {}
|
||||
process = FakeProcess(stdout_lines=["line-1\n", "line-2\n"])
|
||||
|
||||
async def fake_create_subprocess_exec(*args: Any, **kwargs: Any) -> FakeProcess:
|
||||
captured["args"] = args
|
||||
captured["kwargs"] = kwargs
|
||||
return process
|
||||
|
||||
monkeypatch.setattr(exec_module.asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
|
||||
|
||||
exec_client = exec_module.CodexExec(executable_path="/bin/codex", env={"FOO": "bar"})
|
||||
args = exec_module.CodexExecArgs(
|
||||
input="hello",
|
||||
base_url="https://example.com",
|
||||
api_key="api-key",
|
||||
thread_id="thread-123",
|
||||
images=["/tmp/img.png"],
|
||||
model="gpt-4.1-mini",
|
||||
sandbox_mode="read-only",
|
||||
working_directory="/work",
|
||||
additional_directories=["/extra-a", "/extra-b"],
|
||||
skip_git_repo_check=True,
|
||||
output_schema_file="/tmp/schema.json",
|
||||
model_reasoning_effort="high",
|
||||
network_access_enabled=True,
|
||||
web_search_mode="live",
|
||||
approval_policy="on-request",
|
||||
)
|
||||
|
||||
output = [line async for line in exec_client.run(args)]
|
||||
|
||||
assert output == ["line-1", "line-2"]
|
||||
assert process.stdin is not None
|
||||
assert process.stdin.buffer == b"hello"
|
||||
assert process.stdin.closed is True
|
||||
|
||||
assert captured["args"][0] == "/bin/codex"
|
||||
assert list(captured["args"][1:]) == [
|
||||
"exec",
|
||||
"--experimental-json",
|
||||
"--model",
|
||||
"gpt-4.1-mini",
|
||||
"--sandbox",
|
||||
"read-only",
|
||||
"--cd",
|
||||
"/work",
|
||||
"--add-dir",
|
||||
"/extra-a",
|
||||
"--add-dir",
|
||||
"/extra-b",
|
||||
"--skip-git-repo-check",
|
||||
"--output-schema",
|
||||
"/tmp/schema.json",
|
||||
"--config",
|
||||
'model_reasoning_effort="high"',
|
||||
"--config",
|
||||
"sandbox_workspace_write.network_access=true",
|
||||
"--config",
|
||||
'web_search="live"',
|
||||
"--config",
|
||||
'approval_policy="on-request"',
|
||||
"resume",
|
||||
"thread-123",
|
||||
"--image",
|
||||
"/tmp/img.png",
|
||||
"-",
|
||||
]
|
||||
|
||||
env = captured["kwargs"]["env"]
|
||||
assert env["FOO"] == "bar"
|
||||
assert env[exec_module._INTERNAL_ORIGINATOR_ENV] == exec_module._TYPESCRIPT_SDK_ORIGINATOR
|
||||
assert env["OPENAI_BASE_URL"] == "https://example.com"
|
||||
assert env["CODEX_API_KEY"] == "api-key"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("enabled", "expected_config"),
|
||||
[
|
||||
(True, 'web_search="live"'),
|
||||
(False, 'web_search="disabled"'),
|
||||
],
|
||||
)
|
||||
async def test_codex_exec_run_web_search_enabled_flags(
|
||||
monkeypatch: pytest.MonkeyPatch, enabled: bool, expected_config: str
|
||||
) -> None:
|
||||
captured: dict[str, Any] = {}
|
||||
process = FakeProcess(stdout_lines=[])
|
||||
|
||||
async def fake_create_subprocess_exec(*args: Any, **kwargs: Any) -> FakeProcess:
|
||||
captured["args"] = args
|
||||
return process
|
||||
|
||||
monkeypatch.setattr(exec_module.asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
|
||||
|
||||
exec_client = exec_module.CodexExec(executable_path="/bin/codex")
|
||||
args = exec_module.CodexExecArgs(input="hello", web_search_enabled=enabled)
|
||||
|
||||
_ = [line async for line in exec_client.run(args)]
|
||||
command_args = list(captured["args"][1:])
|
||||
assert "--config" in command_args
|
||||
assert expected_config in command_args
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_codex_exec_run_raises_on_non_zero_exit(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
process = FakeProcess(stdout_lines=[], stderr_chunks=[b"bad"], returncode=2)
|
||||
|
||||
async def fake_create_subprocess_exec(*args: Any, **kwargs: Any) -> FakeProcess:
|
||||
return process
|
||||
|
||||
monkeypatch.setattr(exec_module.asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
|
||||
|
||||
exec_client = exec_module.CodexExec(executable_path="/bin/codex")
|
||||
args = exec_module.CodexExecArgs(input="hello")
|
||||
|
||||
with pytest.raises(RuntimeError, match="exited with code 2"):
|
||||
async for _ in exec_client.run(args):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_codex_exec_run_raises_without_stdin(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
process = FakeProcess(stdout_lines=[], stdin_present=False)
|
||||
|
||||
async def fake_create_subprocess_exec(*args: Any, **kwargs: Any) -> FakeProcess:
|
||||
return process
|
||||
|
||||
monkeypatch.setattr(exec_module.asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
|
||||
|
||||
exec_client = exec_module.CodexExec(executable_path="/bin/codex")
|
||||
args = exec_module.CodexExecArgs(input="hello")
|
||||
|
||||
with pytest.raises(RuntimeError, match="no stdin"):
|
||||
async for _ in exec_client.run(args):
|
||||
pass
|
||||
assert process.killed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_codex_exec_run_raises_without_stdout(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
process = FakeProcess(stdout_lines=[], stdout_present=False)
|
||||
|
||||
async def fake_create_subprocess_exec(*args: Any, **kwargs: Any) -> FakeProcess:
|
||||
return process
|
||||
|
||||
monkeypatch.setattr(exec_module.asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
|
||||
|
||||
exec_client = exec_module.CodexExec(executable_path="/bin/codex")
|
||||
args = exec_module.CodexExecArgs(input="hello")
|
||||
|
||||
with pytest.raises(RuntimeError, match="no stdout"):
|
||||
async for _ in exec_client.run(args):
|
||||
pass
|
||||
assert process.killed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_watch_signal_terminates_process() -> None:
|
||||
signal = asyncio.Event()
|
||||
process = FakeProcess(stdout_lines=[], returncode=None)
|
||||
|
||||
task = asyncio.create_task(exec_module._watch_signal(signal, process))
|
||||
signal.set()
|
||||
await task
|
||||
|
||||
assert process.terminated is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("system", "arch", "expected"),
|
||||
[
|
||||
("linux", "x86_64", "x86_64-unknown-linux-musl"),
|
||||
("linux", "aarch64", "aarch64-unknown-linux-musl"),
|
||||
("darwin", "x86_64", "x86_64-apple-darwin"),
|
||||
("darwin", "arm64", "aarch64-apple-darwin"),
|
||||
("win32", "x86_64", "x86_64-pc-windows-msvc"),
|
||||
("win32", "arm64", "aarch64-pc-windows-msvc"),
|
||||
],
|
||||
)
|
||||
def test_platform_target_triple_mapping(
|
||||
monkeypatch: pytest.MonkeyPatch, system: str, arch: str, expected: str
|
||||
) -> None:
|
||||
monkeypatch.setattr(exec_module.sys, "platform", system)
|
||||
monkeypatch.setattr(exec_module.platform, "machine", lambda: arch)
|
||||
assert exec_module._platform_target_triple() == expected
|
||||
|
||||
|
||||
def test_platform_target_triple_unsupported(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(exec_module.sys, "platform", "solaris")
|
||||
monkeypatch.setattr(exec_module.platform, "machine", lambda: "sparc")
|
||||
with pytest.raises(RuntimeError, match="Unsupported platform"):
|
||||
exec_module._platform_target_triple()
|
||||
|
||||
|
||||
def test_find_codex_path_env_override(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("CODEX_PATH", "/custom/codex")
|
||||
assert exec_module.find_codex_path() == "/custom/codex"
|
||||
|
||||
|
||||
def test_find_codex_path_uses_shutil_which(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("CODEX_PATH", raising=False)
|
||||
monkeypatch.setattr(exec_module.shutil, "which", lambda _name: "/usr/local/bin/codex")
|
||||
assert exec_module.find_codex_path() == "/usr/local/bin/codex"
|
||||
|
||||
|
||||
def test_find_codex_path_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("CODEX_PATH", raising=False)
|
||||
monkeypatch.setattr(exec_module.shutil, "which", lambda _name: None)
|
||||
monkeypatch.setattr(exec_module, "_platform_target_triple", lambda: "dummy-triple")
|
||||
monkeypatch.setattr(exec_module.sys, "platform", "linux")
|
||||
result = exec_module.find_codex_path()
|
||||
expected_root = (
|
||||
Path(cast(str, exec_module.__file__)).resolve().parent.parent.parent
|
||||
/ "vendor"
|
||||
/ "dummy-triple"
|
||||
/ "codex"
|
||||
/ "codex"
|
||||
)
|
||||
assert result == str(expected_root)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_thread_run_streamed_passes_options_and_updates_id(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
events = [
|
||||
{"type": "thread.started", "thread_id": "thread-42"},
|
||||
{
|
||||
"type": "turn.completed",
|
||||
"usage": {"input_tokens": 1, "cached_input_tokens": 0, "output_tokens": 1},
|
||||
},
|
||||
]
|
||||
fake_exec = FakeExec(events)
|
||||
options = CodexOptions(base_url="https://example.com", api_key="api-key")
|
||||
thread_options = ThreadOptions(
|
||||
model="gpt-4.1-mini",
|
||||
sandbox_mode="read-only",
|
||||
working_directory="/work",
|
||||
skip_git_repo_check=True,
|
||||
model_reasoning_effort="low",
|
||||
network_access_enabled=False,
|
||||
web_search_mode="cached",
|
||||
approval_policy="on-request",
|
||||
additional_directories=["/extra"],
|
||||
)
|
||||
thread = Thread(
|
||||
exec_client=cast(CodexExec, fake_exec),
|
||||
options=options,
|
||||
thread_options=thread_options,
|
||||
)
|
||||
cleanup_called = False
|
||||
|
||||
def fake_create_output_schema_file(schema: dict[str, Any] | None) -> OutputSchemaFile:
|
||||
nonlocal cleanup_called
|
||||
|
||||
def cleanup() -> None:
|
||||
nonlocal cleanup_called
|
||||
cleanup_called = True
|
||||
|
||||
return OutputSchemaFile(schema_path="/tmp/schema.json", cleanup=cleanup)
|
||||
|
||||
monkeypatch.setattr(thread_module, "create_output_schema_file", fake_create_output_schema_file)
|
||||
|
||||
streamed = await thread.run_streamed(
|
||||
[
|
||||
{"type": "text", "text": "hello"},
|
||||
{"type": "local_image", "path": "/tmp/a.png"},
|
||||
],
|
||||
TurnOptions(output_schema={"type": "object"}),
|
||||
)
|
||||
collected = [event async for event in streamed.events]
|
||||
|
||||
assert collected[0].type == "thread.started"
|
||||
assert thread.id == "thread-42"
|
||||
assert cleanup_called is True
|
||||
|
||||
assert fake_exec.last_args is not None
|
||||
assert fake_exec.last_args.output_schema_file == "/tmp/schema.json"
|
||||
assert fake_exec.last_args.model == "gpt-4.1-mini"
|
||||
assert fake_exec.last_args.sandbox_mode == "read-only"
|
||||
assert fake_exec.last_args.working_directory == "/work"
|
||||
assert fake_exec.last_args.skip_git_repo_check is True
|
||||
assert fake_exec.last_args.model_reasoning_effort == "low"
|
||||
assert fake_exec.last_args.network_access_enabled is False
|
||||
assert fake_exec.last_args.web_search_mode == "cached"
|
||||
assert fake_exec.last_args.approval_policy == "on-request"
|
||||
assert fake_exec.last_args.additional_directories == ["/extra"]
|
||||
assert fake_exec.last_args.images == ["/tmp/a.png"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_thread_run_aggregates_items_and_usage() -> None:
|
||||
events = [
|
||||
{"type": "thread.started", "thread_id": "thread-1"},
|
||||
{
|
||||
"type": "item.completed",
|
||||
"item": {"id": "agent-1", "type": "agent_message", "text": "done"},
|
||||
},
|
||||
{
|
||||
"type": "turn.completed",
|
||||
"usage": {"input_tokens": 2, "cached_input_tokens": 1, "output_tokens": 3},
|
||||
},
|
||||
]
|
||||
thread = Thread(
|
||||
exec_client=cast(CodexExec, FakeExec(events)),
|
||||
options=CodexOptions(),
|
||||
thread_options=ThreadOptions(),
|
||||
)
|
||||
result = await thread.run("hello")
|
||||
|
||||
assert result.final_response == "done"
|
||||
assert result.usage == Usage(
|
||||
input_tokens=2,
|
||||
cached_input_tokens=1,
|
||||
output_tokens=3,
|
||||
)
|
||||
assert len(result.items) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_thread_run_raises_on_failure() -> None:
|
||||
events = [
|
||||
{"type": "turn.failed", "error": {"message": "boom"}},
|
||||
]
|
||||
thread = Thread(
|
||||
exec_client=cast(CodexExec, FakeExec(events)),
|
||||
options=CodexOptions(),
|
||||
thread_options=ThreadOptions(),
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="boom"):
|
||||
await thread.run("hello")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_thread_run_raises_on_stream_error() -> None:
|
||||
events = [
|
||||
{"type": "error", "message": "boom"},
|
||||
]
|
||||
thread = Thread(
|
||||
exec_client=cast(CodexExec, FakeExec(events)),
|
||||
options=CodexOptions(),
|
||||
thread_options=ThreadOptions(),
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="Codex stream error: boom"):
|
||||
await thread.run("hello")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_thread_run_streamed_raises_on_parse_error(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
events = ["not-json"]
|
||||
fake_exec = FakeExec(events)
|
||||
thread = Thread(
|
||||
exec_client=cast(CodexExec, fake_exec),
|
||||
options=CodexOptions(),
|
||||
thread_options=ThreadOptions(),
|
||||
)
|
||||
|
||||
def fake_create_output_schema_file(schema: dict[str, Any] | None) -> OutputSchemaFile:
|
||||
return OutputSchemaFile(schema_path=None, cleanup=lambda: None)
|
||||
|
||||
monkeypatch.setattr(thread_module, "create_output_schema_file", fake_create_output_schema_file)
|
||||
|
||||
streamed = await thread.run_streamed("hello")
|
||||
with pytest.raises(RuntimeError, match="Failed to parse event"):
|
||||
async for _ in streamed.events:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_thread_run_streamed_idle_timeout_sets_signal(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
events = [
|
||||
{
|
||||
"type": "turn.completed",
|
||||
"usage": {"input_tokens": 1, "cached_input_tokens": 0, "output_tokens": 1},
|
||||
}
|
||||
]
|
||||
fake_exec = FakeExec(events, delay=0.2)
|
||||
thread = Thread(
|
||||
exec_client=cast(CodexExec, fake_exec),
|
||||
options=CodexOptions(),
|
||||
thread_options=ThreadOptions(),
|
||||
)
|
||||
signal = asyncio.Event()
|
||||
|
||||
def fake_create_output_schema_file(schema: dict[str, Any] | None) -> OutputSchemaFile:
|
||||
return OutputSchemaFile(schema_path=None, cleanup=lambda: None)
|
||||
|
||||
monkeypatch.setattr(thread_module, "create_output_schema_file", fake_create_output_schema_file)
|
||||
|
||||
with pytest.raises(RuntimeError, match="Codex stream idle for"):
|
||||
async for _ in thread._run_streamed_internal(
|
||||
"hello", TurnOptions(signal=signal, idle_timeout_seconds=0.01)
|
||||
):
|
||||
pass
|
||||
|
||||
assert signal.is_set() is True
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user