wehub/omnigent-ai--omnigent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
omnigent-ai--omnigent/tests/inner/test_codex_executor.py
T
wehub-resource-sync 3de48288cb chore: import upstream snapshot with attribution
2026-07-13 13:12:00 +08:00

2822 lines
105 KiB
Python
Raw Permalink Blame History

"""Tests for CodexExecutor."""
import asyncio
import base64
import json
import tempfile
import unittest
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from omnigent.inner.codex_executor import (
_TURN_EVENT_WARN_SECONDS,
CodexExecutor,
_build_initial_prompt,
_codex_cli_version,
_CodexAppServerSession,
_databricks_codex_config_overrides,
_dynamic_tool_result_payload,
_prompt_for_turn,
_to_codex_input_items,
)
from omnigent.inner.databricks_executor import DatabricksCredentials
from omnigent.inner.executor import (
ExecutorError,
ReasoningChunk,
TextChunk,
ToolCallComplete,
ToolCallRequest,
ToolCallStatus,
TurnComplete,
)
def _run(coro):
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(coro)
finally:
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
@dataclass
class _FakeAppSession:
scripted_turns: list[list[object]]
closed: bool = False
interrupted: bool = False
calls: list[dict] = None # type: ignore[assignment]
def __post_init__(self):
self.calls = []
async def run_turn(self, **kwargs):
self.calls.append(kwargs)
for event in self.scripted_turns.pop(0):
yield event
async def close(self):
self.closed = True
async def interrupt_turn(self):
self.interrupted = True
return True
class _FakePipe:
def __init__(self):
self.writes: list[bytes] = []
self.closed = False
def write(self, data: bytes) -> None:
self.writes.append(data)
async def drain(self) -> None:
return None
def close(self) -> None:
self.closed = True
async def wait_closed(self) -> None:
return None
async def readline(self) -> bytes:
return b""
async def read(self, n: int) -> bytes:
return b""
class _OverflowingPipe:
def __init__(self):
self.read_calls = 0
async def read(self, n: int) -> bytes:
if self.read_calls == 0:
self.read_calls += 1
return (b"x" * n) + b"\n"
return b""
class _ChunkedPipe:
def __init__(self, chunks: list[bytes]):
self._chunks = list(chunks)
async def read(self, n: int) -> bytes:
if not self._chunks:
return b""
return self._chunks.pop(0)
class _FakeProcess:
def __init__(self):
self.stdin = _FakePipe()
self.stdout = _FakePipe()
self.stderr = _FakePipe()
self.returncode: int | None = None
self.terminated = False
self.pid = 12345
def terminate(self) -> None:
self.terminated = True
self.returncode = 0
def kill(self) -> None:
self.returncode = -9
async def wait(self) -> int:
return self.returncode or 0
class TestCodexExecutor(unittest.TestCase):
def test_databricks_codex_config_overrides(self):
overrides = _databricks_codex_config_overrides(
model="databricks-gpt-5-4-mini",
base_url="https://example.cloud.databricks.com/ai-gateway/codex/v1",
auth_command=('databricks auth token --host "https://example.cloud.databricks.com"'),
)
self.assertIn('model="databricks-gpt-5-4-mini"', overrides)
self.assertIn('model_provider="omnigent_databricks"', overrides)
self.assertTrue(any("/ai-gateway/codex/v1" in item for item in overrides))
self.assertFalse(any("/serving-endpoints" in item for item in overrides))
self.assertTrue(any('auth={command="sh"' in item for item in overrides))
self.assertTrue(any("databricks auth token --host" in item for item in overrides))
self.assertTrue(any("refresh_interval_ms=900000" in item for item in overrides))
self.assertFalse(any('env_key="DATABRICKS_TOKEN"' in item for item in overrides))
def test_codex_config_overrides_neutralize_toml_breakout(self):
"""A model id full of TOML metacharacters stays a literal string.
Defense-in-depth for the model_override RCE: the model value is
``json.dumps``-escaped, so even a string crafted to close the
``model="..."`` field and inject its own ``auth.command`` parses
back as one inert model name and never overwrites the real
token-minting auth command.
"""
import tomllib
payload = 'x",auth={command="sh",args=["-c","touch /tmp/pwned"]},wire_api="responses"}'
real_auth = 'databricks auth token --host "https://example.cloud.databricks.com"'
overrides = _databricks_codex_config_overrides(
model=payload,
base_url="https://example.cloud.databricks.com/ai-gateway/codex/v1",
auth_command=real_auth,
)
parsed = tomllib.loads("\n".join(overrides))
# The whole payload round-trips as the literal model name.
self.assertEqual(parsed["model"], payload)
# The injected auth command did not survive — the legit one did.
provider = parsed["model_providers"]["omnigent_databricks"]
self.assertEqual(provider["auth"]["args"], ["-c", real_auth])
def test_constructor_databricks_flag_with_profile(self):
with (
patch("omnigent.inner.codex_executor._find_codex_cli", return_value="/usr/bin/codex"),
patch(
"omnigent.inner.codex_executor._read_databrickscfg",
return_value=DatabricksCredentials(
host="https://example.cloud.databricks.com",
token="dapi_test_token",
),
),
):
executor = CodexExecutor(gateway=True)
self.assertTrue(executor._gateway)
self.assertEqual(
executor._env["DATABRICKS_HOST"],
"https://example.cloud.databricks.com",
)
self.assertNotIn("DATABRICKS_TOKEN", executor._env)
self.assertIn("model=", executor._codex_config_overrides[0])
self.assertIn('model_provider="omnigent_databricks"', executor._codex_config_overrides[1])
def test_constructor_does_not_force_codex_debug_env_by_default(self):
with (
patch("omnigent.inner.codex_executor._find_codex_cli", return_value="/usr/bin/codex"),
patch.dict("os.environ", {}, clear=True),
):
executor = CodexExecutor()
self.assertNotIn("RUST_LOG", executor._env)
self.assertNotIn("RUST_BACKTRACE", executor._env)
def test_constructor_databricks_flag_with_profile_uses_profile_credentials(self):
with (
patch("omnigent.inner.codex_executor._find_codex_cli", return_value="/usr/bin/codex"),
patch.dict("os.environ", {}, clear=True),
patch(
"omnigent.inner.codex_executor._read_databrickscfg",
return_value=DatabricksCredentials(
host="https://example-profile-workspace.cloud.databricks.com",
token="profile_token",
),
),
):
executor = CodexExecutor(
gateway=True,
databricks_profile="test-profile",
model="databricks-gpt-5-4-mini",
)
self.assertEqual(
executor._env["DATABRICKS_HOST"],
"https://example-profile-workspace.cloud.databricks.com",
)
self.assertNotIn("DATABRICKS_TOKEN", executor._env)
# The fix: with an explicit profile, the bearer-token helper selects by
# --profile (unambiguous), never --host. A regression to --host makes a
# workspace with two profiles on one host return an empty token → 401.
self.assertTrue(
any(
"databricks auth token --profile" in override
for override in executor._codex_config_overrides
)
)
self.assertFalse(
any("--host" in override for override in executor._codex_config_overrides)
)
# `--force-refresh` only exists in Databricks CLI >= v0.296.0, so it
# must be applied via a `--help` capability probe ($force), never
# passed unconditionally — an older CLI rejects the unknown flag and
# yields an empty token → silent 401.
auth_override = next(
o for o in executor._codex_config_overrides if "databricks auth token" in o
)
self.assertIn("databricks auth token --help", auth_override)
self.assertIn("force=--force-refresh", auth_override)
self.assertNotIn('--profile "test-profile" --force-refresh', auth_override)
def test_constructor_databricks_flag_with_host_override_skips_profile_lookup(self):
with (
patch("omnigent.inner.codex_executor._find_codex_cli", return_value="/usr/bin/codex"),
patch.dict("os.environ", {}, clear=True),
patch("omnigent.inner.codex_executor._read_databrickscfg") as read_cfg,
):
executor = CodexExecutor(
gateway=True,
databricks_profile="missing-profile",
gateway_host="https://example.databricks.com/",
base_url_override="https://example.databricks.com/ai-gateway/codex/v1",
gateway_auth_command="printf token",
model="databricks-gpt-5-4-mini",
)
read_cfg.assert_not_called()
self.assertEqual(
executor._env["DATABRICKS_HOST"],
"https://example.databricks.com",
)
self.assertNotIn("DATABRICKS_TOKEN", executor._env)
self.assertTrue(
any("printf token" in override for override in executor._codex_config_overrides)
)
self.assertFalse(
any(
"databricks auth token --host" in item for item in executor._codex_config_overrides
)
)
def test_constructor_databricks_flag_with_host_override_requires_base_url(self):
with (
patch("omnigent.inner.codex_executor._find_codex_cli", return_value="/usr/bin/codex"),
patch.dict("os.environ", {}, clear=True),
self.assertRaisesRegex(OSError, "GATEWAY_BASE_URL"),
):
CodexExecutor(
gateway=True,
gateway_host="https://example.databricks.com/",
gateway_auth_command="printf token",
)
def test_constructor_databricks_flag_with_host_override_requires_auth_command(self):
with (
patch("omnigent.inner.codex_executor._find_codex_cli", return_value="/usr/bin/codex"),
patch.dict("os.environ", {}, clear=True),
self.assertRaisesRegex(OSError, "GATEWAY_AUTH_COMMAND"),
):
CodexExecutor(
gateway=True,
gateway_host="https://example.databricks.com/",
base_url_override="https://example.databricks.com/ai-gateway/codex/v1",
)
def test_constructor_databricks_flag_no_creds_raises(self):
with (
patch("omnigent.inner.codex_executor._find_codex_cli", return_value="/usr/bin/codex"),
patch.dict("os.environ", {}, clear=True),
patch("omnigent.inner.codex_executor._read_databrickscfg", return_value=None),
patch("omnigent.inner.codex_executor._read_databrickscfg_host", return_value=None),
):
with self.assertRaises(EnvironmentError):
CodexExecutor(gateway=True)
def test_build_initial_prompt_serializes_history(self):
prompt = _build_initial_prompt(
[
{"role": "user", "content": "one"},
{"role": "assistant", "content": "two"},
{"role": "user", "content": "three"},
]
)
self.assertIn("Conversation so far:", prompt)
self.assertIn("assistant: two", prompt)
self.assertIn("user: three", prompt)
def test_dynamic_tool_result_payload_uses_content_items(self):
payload = _dynamic_tool_result_payload({"answer": 5})
self.assertEqual(payload["success"], True)
self.assertEqual(payload["contentItems"][0]["type"], "inputText")
self.assertIn('"answer": 5', payload["contentItems"][0]["text"])
def test_prompt_for_new_thread_keeps_history(self):
messages = [
{"role": "user", "content": "The secret codeword is ZEBRA-99."},
{"role": "assistant", "content": "Understood."},
{"role": "user", "content": "Summarize our conversation."},
]
prompt = _prompt_for_turn(messages, is_new_thread=True)
self.assertIn("Conversation so far:", prompt)
self.assertIn("ZEBRA-99", prompt)
self.assertIn("Summarize our conversation.", prompt)
def test_prompt_for_existing_thread_uses_latest_user_message(self):
messages = [
{"role": "user", "content": "The secret codeword is ZEBRA-99."},
{"role": "assistant", "content": "Understood."},
{"role": "user", "content": "Summarize our conversation."},
]
prompt = _prompt_for_turn(messages, is_new_thread=False)
self.assertEqual(prompt, "Summarize our conversation.")
def test_run_turn_delegates_to_app_server_session(self):
async def _t():
fake_session = _FakeAppSession(
[
[
TextChunk(text="Checking..."),
ToolCallRequest(name="calculate", args={"expression": "2+3"}),
ToolCallComplete(
name="calculate",
status=ToolCallStatus.SUCCESS,
result={"result": 5},
),
TurnComplete(response="5"),
]
]
)
executor = CodexExecutor(
codex_path="/bin/echo",
app_session_factory=lambda **kwargs: fake_session,
)
events = [
e
async for e in executor.run_turn(
[{"role": "user", "content": "calc", "session_id": "s1"}],
[
{
"name": "calculate",
"description": "Calculate",
"parameters": {
"type": "object",
"properties": {"expression": {"type": "string"}},
},
}
],
"Be helpful.",
)
]
self.assertIsInstance(events[0], TextChunk)
self.assertIsInstance(events[1], ToolCallRequest)
self.assertIsInstance(events[2], ToolCallComplete)
self.assertIsInstance(events[3], TurnComplete)
self.assertEqual(fake_session.calls[0]["system_prompt"], "Be helpful.")
self.assertEqual(fake_session.calls[0]["model"], "gpt-5.4-mini")
self.assertEqual(fake_session.calls[0]["tools"][0]["name"], "calculate")
_run(_t())
def test_run_turn_databricks_uses_databricks_default_model(self):
async def _t():
fake_session = _FakeAppSession([[TurnComplete(response="done")]])
with patch(
"omnigent.inner.codex_executor._read_databrickscfg",
return_value=DatabricksCredentials(
host="https://example.cloud.databricks.com",
token="dapi_test_token",
),
):
executor = CodexExecutor(
codex_path="/bin/echo",
gateway=True,
app_session_factory=lambda **kwargs: fake_session,
)
events = [
e
async for e in executor.run_turn(
[{"role": "user", "content": "hi", "session_id": "s1"}],
[],
"",
)
]
self.assertEqual(events[-1].response, "done")
self.assertEqual(fake_session.calls[0]["model"], "databricks-gpt-5-5")
_run(_t())
def test_run_turn_passes_session_send_tool_through(self):
async def _t():
fake_session = _FakeAppSession([[TurnComplete(response="done")]])
executor = CodexExecutor(
codex_path="/bin/echo",
app_session_factory=lambda **kwargs: fake_session,
)
events = [
e
async for e in executor.run_turn(
[{"role": "user", "content": "hi", "session_id": "s1"}],
[
{
"name": "sys_session_send",
"description": "Session send",
"parameters": {
"type": "object",
"properties": {
"tool": {"type": "string"},
"session": {"type": "string"},
"args": {"type": "object"},
},
},
}
],
"Be helpful.",
)
]
self.assertEqual(fake_session.calls[0]["tools"][0]["name"], "sys_session_send")
self.assertIsInstance(events[-1], TurnComplete)
_run(_t())
def test_app_server_run_turn_keeps_native_shell_by_default(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session._request = AsyncMock(
side_effect=[
{"result": {"thread": {"id": "thread-1"}}},
{"result": {"turn": {"id": "turn-1"}}},
[
{
"method": "turn/completed",
"params": {
"turn": {"id": "turn-1"},
},
}
][0],
]
)
async def _inject_turn_completed() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "turn/completed",
"params": {
"turn": {"id": "turn-1"},
},
}
)
inject_task = asyncio.create_task(_inject_turn_completed())
_ = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[
{
"name": "sys_os_shell",
"description": "Shell",
"parameters": {
"type": "object",
"properties": {"command": {"type": "string"}},
},
}
],
system_prompt="Be helpful.",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
thread_start_call = session._request.await_args_list[0]
params = thread_start_call.args[1]
self.assertNotIn("shell_tool", params["config"]["features"])
self.assertEqual(params["dynamicTools"][0]["name"], "sys_os_shell")
_run(_t())
def test_app_server_run_turn_applies_effort_via_thread_settings_update(self):
"""Reasoning effort rides thread/settings/update, not turn/start.
Codex's ``TurnStartParams`` has no ``effort`` field, so an effort set
on ``turn/start`` is silently dropped by serde and never takes effect.
It must go through ``thread/settings/update`` (whose
``ThreadSettingsUpdateParams`` carries ``effort``) — the same path the
TUI ``/model`` picker uses.
"""
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session._request = AsyncMock(
side_effect=[
{"result": {"thread": {"id": "thread-1"}}}, # thread/start
{"result": {}}, # thread/settings/update
{"result": {"turn": {"id": "turn-1"}}}, # turn/start
]
)
async def _inject_turn_completed() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{"method": "turn/completed", "params": {"turn": {"id": "turn-1"}}}
)
inject_task = asyncio.create_task(_inject_turn_completed())
# Drive the turn to completion (consume the event stream for its
# side effects — the RPCs we assert on below).
async for _event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
reasoning_effort="high",
):
pass
await inject_task
methods = [call.args[0] for call in session._request.await_args_list]
# Settings update lands BEFORE the turn starts so the effort applies
# to this turn, and turn/start carries no (dropped) effort field.
self.assertEqual(methods, ["thread/start", "thread/settings/update", "turn/start"])
settings_params = session._request.await_args_list[1].args[1]
self.assertEqual(settings_params, {"threadId": "thread-1", "effort": "high"})
turn_params = session._request.await_args_list[2].args[1]
self.assertNotIn("effort", turn_params)
_run(_t())
def test_app_server_run_turn_dedupes_unchanged_effort(self):
"""An unchanged effort is not re-sent on a later turn of one thread.
Effort persists on the thread once applied, so re-issuing
``thread/settings/update`` every turn would be a redundant RPC.
"""
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
# Existing thread (no thread/start) with effort already applied.
session.thread_id = "thread-1"
session._applied_effort = "high"
session._request = AsyncMock(side_effect=[{"result": {"turn": {"id": "turn-2"}}}])
async def _inject_turn_completed() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{"method": "turn/completed", "params": {"turn": {"id": "turn-2"}}}
)
inject_task = asyncio.create_task(_inject_turn_completed())
# Drive the turn to completion (consume the event stream for its
# side effects — the RPCs we assert on below).
async for _event in session.run_turn(
messages=[{"role": "user", "content": "again"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
reasoning_effort="high",
):
pass
await inject_task
methods = [call.args[0] for call in session._request.await_args_list]
self.assertEqual(methods, ["turn/start"])
_run(_t())
def test_app_server_dynamic_tool_complete_carries_call_id_metadata(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=lambda name, args: {"ok": True, "name": name, "args": args},
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
async def _inject_events() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"id": 7,
"method": "item/tool/call",
"params": {
"turnId": "turn-1",
"callId": "call_abc123",
"tool": "sys_os_shell",
"arguments": {"command": "pwd"},
},
}
)
session._events.put_nowait(
{
"method": "turn/completed",
"params": {"turn": {"id": "turn-1"}},
}
)
inject_task = asyncio.create_task(_inject_events())
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "pwd"}],
tools=[{"name": "sys_os_shell", "description": "Shell"}],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
tool_request = next(e for e in events if isinstance(e, ToolCallRequest))
tool_complete = next(e for e in events if isinstance(e, ToolCallComplete))
self.assertEqual(tool_request.metadata, {"call_id": "call_abc123"})
self.assertEqual(tool_complete.metadata, {"call_id": "call_abc123"})
_run(_t())
def test_app_server_run_turn_can_disable_native_tools_when_requested(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
disable_native_tools=True,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session._request = AsyncMock(
side_effect=[
{"result": {"thread": {"id": "thread-1"}}},
{"result": {"turn": {"id": "turn-1"}}},
]
)
async def _inject_turn_completed() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "turn/completed",
"params": {
"turn": {"id": "turn-1"},
},
}
)
inject_task = asyncio.create_task(_inject_turn_completed())
_ = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[
{
"name": "sys_os_shell",
"description": "Shell",
"parameters": {
"type": "object",
"properties": {"command": {"type": "string"}},
},
}
],
system_prompt="Be helpful.",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
thread_start_call = session._request.await_args_list[0]
params = thread_start_call.args[1]
self.assertEqual(params["config"]["features"]["shell_tool"], False)
_run(_t())
def test_close_uses_process_tree_termination(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session._proc = _FakeProcess()
session._started = True
with patch("omnigent.inner.codex_executor._terminate_process_tree") as terminate_tree:
await session.close()
terminate_tree.assert_called_once()
_run(_t())
def test_signature_change_recreates_app_session(self):
async def _t():
first = _FakeAppSession([[TurnComplete(response="one")]])
second = _FakeAppSession([[TurnComplete(response="two")]])
sessions = [first, second]
executor = CodexExecutor(
codex_path="/bin/echo",
app_session_factory=lambda **kwargs: sessions.pop(0),
)
first_events = [
e
async for e in executor.run_turn(
[{"role": "user", "content": "hi", "session_id": "s1"}],
[],
"",
)
]
second_events = [
e
async for e in executor.run_turn(
[{"role": "user", "content": "hi", "session_id": "s1"}],
[
{
"name": "sleep",
"description": "Sleep",
"parameters": {"type": "object", "properties": {}},
}
],
"",
)
]
self.assertEqual(first_events[-1].response, "one")
self.assertEqual(second_events[-1].response, "two")
self.assertTrue(first.closed)
self.assertFalse(second.closed)
_run(_t())
def test_close_session_closes_app_session(self):
async def _t():
fake_session = _FakeAppSession([[TurnComplete(response="done")]])
executor = CodexExecutor(
codex_path="/bin/echo",
app_session_factory=lambda **kwargs: fake_session,
)
_ = [
e
async for e in executor.run_turn(
[{"role": "user", "content": "hi", "session_id": "s1"}],
[],
"",
)
]
await executor.close_session("s1")
self.assertTrue(fake_session.closed)
_run(_t())
def test_enqueue_session_message_steers_active_turn(self):
async def _t():
fake_session = _FakeAppSession([[TurnComplete(response="done")]])
fake_session.enqueue_message = AsyncMock(return_value=True)
executor = CodexExecutor(
codex_path="/bin/echo",
app_session_factory=lambda **kwargs: fake_session,
)
_ = [
e
async for e in executor.run_turn(
[{"role": "user", "content": "hi", "session_id": "s1"}],
[],
"",
)
]
queued = await executor.enqueue_session_message("s1", "stop")
self.assertTrue(queued)
fake_session.enqueue_message.assert_awaited_once_with("stop")
self.assertTrue(executor.supports_live_message_queue())
_run(_t())
def test_interrupt_session_interrupts_then_drops_session(self):
"""A user interrupt halts the turn AND drops the session.
Codex resumes the same thread on the next turn and sends only the
latest user message, so a retained session would bypass the
runner's ``[System: interrupted]`` marker and continue the
abandoned request. Dropping the session (close resets thread_id)
forces the next turn to start a fresh thread and replay full
history. An empty ``_session_states`` is the invariant that
prevents the canceled-instruction leak.
"""
async def _t():
fake_session = _FakeAppSession([[TurnComplete(response="done")]])
executor = CodexExecutor(
codex_path="/bin/echo",
app_session_factory=lambda **kwargs: fake_session,
)
_ = [
e
async for e in executor.run_turn(
[{"role": "user", "content": "hi", "session_id": "s1"}],
[],
"",
)
]
interrupted = await executor.interrupt_session("s1")
self.assertTrue(interrupted)
# Interrupt is attempted first to halt the in-flight turn.
self.assertTrue(fake_session.interrupted)
# Then the session is closed and removed so the next turn starts a
# fresh thread and replays full history (marker included).
self.assertTrue(fake_session.closed)
self.assertEqual(executor._session_states, {})
_run(_t())
def test_app_server_uses_workspace_cwd(self):
async def _t():
fake_proc = _FakeProcess()
recorded_cwd: str | None = None
async def _fake_create_subprocess_exec(*args, **kwargs):
nonlocal recorded_cwd
recorded_cwd = kwargs.get("cwd")
return fake_proc
with tempfile.TemporaryDirectory() as workspace:
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd=workspace,
env={},
tool_executor=None,
)
session._request = AsyncMock(return_value={"result": {}})
with patch(
"omnigent.inner.codex_executor._create_subprocess_exec",
new=_fake_create_subprocess_exec,
):
await session.start()
assert recorded_cwd is not None
self.assertEqual(recorded_cwd, workspace)
await session.close()
self.assertTrue(fake_proc.terminated)
_run(_t())
def test_app_server_uses_isolated_codex_home_and_cleans_it_up(self):
"""Codex subprocess must receive a private CODEX_HOME, not ~/.codex/."""
async def _t():
fake_proc = _FakeProcess()
recorded_env: dict | None = None
async def _fake_create_subprocess_exec(*args, **kwargs):
nonlocal recorded_env
recorded_env = kwargs.get("env")
return fake_proc
with tempfile.TemporaryDirectory() as workspace:
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd=workspace,
env={},
tool_executor=None,
)
session._request = AsyncMock(return_value={"result": {}})
with patch(
"omnigent.inner.codex_executor._create_subprocess_exec",
new=_fake_create_subprocess_exec,
):
await session.start()
assert recorded_env is not None
self.assertIn("CODEX_HOME", recorded_env)
codex_home = Path(recorded_env["CODEX_HOME"])
self.assertTrue(codex_home.is_dir())
self.assertTrue(codex_home.name.startswith("omnigent-codex-home-"))
self.assertTrue(str(codex_home).startswith(tempfile.gettempdir()))
# Must not point at the user's real ~/.codex directory.
self.assertNotEqual(codex_home, Path.home() / ".codex")
await session.close()
self.assertFalse(codex_home.exists())
_run(_t())
def test_app_server_enqueue_message_uses_turn_steer(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.thread_id = "thread-1"
session.active_turn_id = "turn-1"
session._request = AsyncMock(return_value={"result": {"turnId": "turn-1"}})
queued = await session.enqueue_message("stop")
self.assertTrue(queued)
session._request.assert_awaited_once_with(
"turn/steer",
{
"threadId": "thread-1",
"expectedTurnId": "turn-1",
"input": [{"type": "text", "text": "stop"}],
},
)
self.assertEqual(session.active_turn_id, "turn-1")
_run(_t())
def test_app_server_interrupt_turn_uses_turn_interrupt(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.thread_id = "thread-1"
session.active_turn_id = "turn-1"
session._request = AsyncMock(return_value={"result": {}})
interrupted = await session.interrupt_turn()
self.assertTrue(interrupted)
session._request.assert_awaited_once_with(
"turn/interrupt",
{
"threadId": "thread-1",
"turnId": "turn-1",
},
)
_run(_t())
def test_app_server_run_turn_warns_and_keeps_waiting_when_events_are_slow(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
original_warn = _TURN_EVENT_WARN_SECONDS
import omnigent.inner.codex_executor as codex_executor_module
codex_executor_module._TURN_EVENT_WARN_SECONDS = 0.05
try:
async def _inject_final_after_warning() -> None:
# Let the warning fire at least once, then deliver a terminal event.
await asyncio.sleep(0.12)
session._events.put_nowait(
{
"method": "item/completed",
"params": {
"turnId": "turn-1",
"item": {
"id": "msg-1",
"type": "agentMessage",
"text": "done",
"phase": "final_answer",
},
},
}
)
inject_task = asyncio.create_task(_inject_final_after_warning())
with self.assertLogs("omnigent.inner.codex_executor", level="WARNING") as cm:
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
finally:
codex_executor_module._TURN_EVENT_WARN_SECONDS = original_warn
self.assertTrue(
any("has been idle for" in line for line in cm.output),
cm.output,
)
self.assertTrue(
any(type(evt).__name__ == "TurnComplete" for evt in events),
events,
)
_run(_t())
def test_app_server_run_turn_finishes_on_final_answer_item_without_turn_completed(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
await session._events.put(
{
"method": "item/completed",
"params": {
"turnId": "turn-1",
"item": {
"id": "msg-1",
"type": "agentMessage",
"phase": "final_answer",
"text": "done",
},
},
}
)
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
self.assertEqual(len(events), 2)
self.assertIsInstance(events[0], TextChunk)
self.assertEqual(events[0].text, "done")
self.assertIsInstance(events[1], TurnComplete)
self.assertEqual(events[1].response, "done")
self.assertIsNone(session.active_turn_id)
_run(_t())
def test_app_server_run_turn_attaches_token_usage_to_turn_complete(self):
"""``thread/tokenUsage/updated`` payloads populate ``TurnComplete.usage``.
Without forwarding the codex token-usage notification, harness-backed
codex tests produce empty token-usage artifacts even though the
provider counted real tokens. This test asserts the inner executor
captures the latest ``tokenUsage.last`` breakdown and attaches it
to the next ``TurnComplete`` it yields.
"""
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
async def _inject_events() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "thread/tokenUsage/updated",
"params": {
"threadId": "thread-1",
"turnId": "turn-1",
"tokenUsage": {
"last": {
"inputTokens": 100,
"outputTokens": 25,
"totalTokens": 125,
"cachedInputTokens": 0,
"reasoningOutputTokens": 0,
},
"total": {
"inputTokens": 100,
"outputTokens": 25,
"totalTokens": 125,
"cachedInputTokens": 0,
"reasoningOutputTokens": 0,
},
},
},
}
)
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "turn/completed",
"params": {"turn": {"id": "turn-1"}},
}
)
inject_task = asyncio.create_task(_inject_events())
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
self.assertEqual(len(events), 1)
self.assertIsInstance(events[0], TurnComplete)
self.assertEqual(
events[0].usage,
{"input_tokens": 100, "output_tokens": 25, "total_tokens": 125},
)
# The cached usage must be cleared after consumption so the next
# turn doesn't inherit stale numbers.
self.assertIsNone(session._last_turn_usage)
_run(_t())
def test_app_server_run_turn_notifies_usage_observer(self):
"""The inner executor's ``TurnComplete`` yield site notifies the
shared usage observer so integration tests that drive the codex
executor directly populate the per-test token-usage artifact."""
from omnigent.llms import _usage_observer
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
seen: list[dict[str, object]] = []
remove = _usage_observer.add_observer(lambda **kw: seen.append(kw))
async def _inject_events() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "thread/tokenUsage/updated",
"params": {
"threadId": "thread-1",
"turnId": "turn-1",
"tokenUsage": {
"last": {
"inputTokens": 100,
"outputTokens": 25,
"totalTokens": 125,
"cachedInputTokens": 0,
"reasoningOutputTokens": 0,
},
"total": {
"inputTokens": 100,
"outputTokens": 25,
"totalTokens": 125,
"cachedInputTokens": 0,
"reasoningOutputTokens": 0,
},
},
},
}
)
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "turn/completed",
"params": {"turn": {"id": "turn-1"}},
}
)
inject_task = asyncio.create_task(_inject_events())
try:
[
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="codex-test-model",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
finally:
remove()
self.assertEqual(
seen,
[
{
"model": "codex-test-model",
"input_tokens": 100,
"output_tokens": 25,
"total_tokens": 125,
}
],
)
_run(_t())
def test_app_server_run_turn_usage_none_when_no_token_notification(self):
"""Without a ``thread/tokenUsage/updated`` event, ``TurnComplete.usage`` is None."""
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
async def _inject_completed() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "turn/completed",
"params": {"turn": {"id": "turn-1"}},
}
)
inject_task = asyncio.create_task(_inject_completed())
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
self.assertEqual(len(events), 1)
self.assertIsInstance(events[0], TurnComplete)
self.assertIsNone(events[0].usage)
_run(_t())
def test_app_server_run_turn_drains_final_answer_after_turn_completed(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
async def _inject_trailing_final_answer() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "turn/completed",
"params": {
"turn": {"id": "turn-1"},
},
}
)
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "item/completed",
"params": {
"turnId": "turn-1",
"item": {
"id": "msg-1",
"type": "agentMessage",
"phase": "final_answer",
"text": "done",
},
},
}
)
inject_task = asyncio.create_task(_inject_trailing_final_answer())
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
self.assertEqual(len(events), 1)
self.assertIsInstance(events[0], TurnComplete)
self.assertEqual(events[0].response, "done")
_run(_t())
def test_app_server_run_turn_uses_buffered_delta_when_completed_item_is_missing(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
await session._events.put(
{
"method": "item/agentMessage/delta",
"params": {
"turnId": "turn-1",
"itemId": "msg-1",
"delta": "done",
},
}
)
await session._events.put(
{
"method": "turn/completed",
"params": {
"turn": {"id": "turn-1"},
},
}
)
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
self.assertEqual(len(events), 1)
self.assertIsInstance(events[0], TurnComplete)
self.assertEqual(events[0].response, "done")
_run(_t())
def test_app_server_run_turn_reasoning_deltas_yield_reasoning_chunks(self):
"""item/reasoning/textDelta and item/reasoning/summaryTextDelta events
yield ReasoningChunk events so the idle watchdog resets during long
think phases (regression guard for omnigent-ai/omnigent#738)."""
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
async def _inject() -> None:
await asyncio.sleep(0.01)
session._events.put_nowait(
{
"method": "item/reasoning/textDelta",
"params": {"turnId": "turn-1", "delta": "thinking hard..."},
}
)
session._events.put_nowait(
{
"method": "item/reasoning/summaryTextDelta",
"params": {"turnId": "turn-1", "delta": "summary of thoughts"},
}
)
session._events.put_nowait(
{
"method": "item/completed",
"params": {
"turnId": "turn-1",
"item": {
"id": "msg-1",
"type": "agentMessage",
"phase": "final_answer",
"text": "Here is my answer.",
},
},
}
)
inject_task = asyncio.create_task(_inject())
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "complex question"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
reasoning_events = [e for e in events if isinstance(e, ReasoningChunk)]
self.assertEqual(len(reasoning_events), 2)
self.assertEqual(reasoning_events[0].delta, "thinking hard...")
self.assertEqual(reasoning_events[0].event_type, "reasoning_text")
self.assertEqual(reasoning_events[1].delta, "summary of thoughts")
self.assertEqual(reasoning_events[1].event_type, "reasoning_text")
turn_complete = events[-1]
self.assertIsInstance(turn_complete, TurnComplete)
self.assertEqual(turn_complete.response, "Here is my answer.")
_run(_t())
def test_stderr_loop_handles_oversized_lines(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
proc = _FakeProcess()
proc.stderr = _OverflowingPipe()
session._proc = proc
await session._stderr_loop()
self.assertEqual(len(session._recent_stderr), 1)
self.assertTrue(session._recent_stderr[0].endswith("...[truncated]"))
_run(_t())
def test_reader_loop_handles_large_json_message_split_across_chunks(self):
async def _t():
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
proc = _FakeProcess()
payload = {
"method": "item/completed",
"params": {
"turnId": "turn-1",
"item": {
"id": "msg-1",
"type": "agentMessage",
"phase": "final_answer",
"text": "x" * 70000,
},
},
}
encoded = (json.dumps(payload) + "\n").encode("utf-8")
proc.stdout = _ChunkedPipe([encoded[:100], encoded[100:50000], encoded[50000:]])
session._proc = proc
reader_task = asyncio.create_task(session._reader_loop())
await reader_task
message = await asyncio.wait_for(session._events.get(), timeout=0.1)
self.assertEqual(message["method"], "item/completed")
self.assertEqual(message["params"]["item"]["phase"], "final_answer")
self.assertEqual(len(message["params"]["item"]["text"]), 70000)
_run(_t())
# ── Retryable ExecutorError emission ──────────────────────────
#
# Function-based pytest tests for the retryable-flag behavior on the
# two codex app-server failure paths. Kept outside the unittest class
# above to comply with the project-wide function-based test rule —
# the class-style tests above pre-date that rule.
async def test_run_turn_turn_failed_emits_retryable_executor_error() -> None:
"""
When the codex app server reports ``method == "turn/failed"`` for
the active turn, ``_CodexAppServerSession.run_turn`` must emit
exactly one ``ExecutorError`` with ``retryable=True``. Without the
retryable flag the workflow raises ``PermanentLLMError`` and the
surrounding retry policy never gets a chance to reissue a turn
that failed for a transient reason (provider hiccup, tool exit
code, etc.).
If this breaks (e.g. someone drops ``retryable=True``), the user
sees transient codex failures as terminal task errors instead of
a transparent retry.
"""
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
# ``turnId`` must be set in ``params`` so the startup drain at
# codex_executor.py:550 preserves the event for the main loop —
# real codex frames carry it alongside the nested ``turn.id``.
await session._events.put(
{
"method": "turn/failed",
"params": {
"turnId": "turn-1",
"turn": {"id": "turn-1", "error": "bash exit 127: printf: illegal usage"},
"message": "bash exit 127: printf: illegal usage",
},
}
)
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
# Exactly one ExecutorError — anything else means the terminal
# event stream leaked extra events past the failure.
error_events = [e for e in events if isinstance(e, ExecutorError)]
assert len(error_events) == 1, (
f"Expected exactly 1 ExecutorError on turn/failed, got {len(error_events)}: {events}"
)
assert error_events[0].retryable is True
assert "bash exit 127" in error_events[0].message
async def test_run_turn_method_error_emits_retryable_executor_error() -> None:
"""
When the codex app server emits a top-level ``method == "error"``
JSON-RPC frame (typical for runtime / tool-execution failures),
``run_turn`` must emit ``ExecutorError(retryable=True)`` so the
workflow's retry policy can reissue. Matches the behavior for
``turn/failed`` — both paths carry transient provider-side errors.
"""
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-1"}}})
# ``turnId`` is included only so the startup drain at
# codex_executor.py:550 preserves the event for the main loop;
# the ``method == error`` handler itself ignores ``turnId``.
await session._events.put(
{
"method": "error",
"params": {"turnId": "turn-1", "message": "shell command crashed"},
}
)
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
error_events = [e for e in events if isinstance(e, ExecutorError)]
assert len(error_events) == 1, (
f"Expected exactly 1 ExecutorError on method=error, got {len(error_events)}: {events}"
)
assert error_events[0].retryable is True
assert "shell command crashed" in error_events[0].message
async def test_run_turn_refuses_to_adopt_stale_final_answer_event() -> None:
"""Adopt fallback must drop a stale final-answer item rather
than adopt it as the new turn's response.
"""
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
# ``turn/start`` reports the new turn id (turn-2). The first event
# the executor sees is a stale final-answer item carrying turn-1's
# id (the prior turn's tail), followed immediately by the real
# turn-2 final-answer.
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-2"}}})
async def _inject_stale_then_real() -> None:
await asyncio.sleep(0.01)
# Stale tail: prior turn's final answer arrives during turn-2.
session._events.put_nowait(
{
"method": "item/completed",
"params": {
"turnId": "turn-1",
"item": {
"type": "agentMessage",
"id": "msg-stale",
"text": "STALE",
"phase": "final_answer",
},
},
}
)
# The real turn-2 final answer.
session._events.put_nowait(
{
"method": "item/completed",
"params": {
"turnId": "turn-2",
"item": {
"type": "agentMessage",
"id": "msg-real",
"text": "REAL",
"phase": "final_answer",
},
},
}
)
inject_task = asyncio.create_task(_inject_stale_then_real())
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
turn_completes = [e for e in events if isinstance(e, TurnComplete)]
assert len(turn_completes) == 1, (
f"Expected exactly one TurnComplete, got {len(turn_completes)}: {events}"
)
# ``STALE`` here means the adopt fallback fired on the stale
# final-answer; the fix is meant to drop it instead.
assert turn_completes[0].response == "REAL", (
f"Expected 'REAL' (real turn-2 response); got "
f"{turn_completes[0].response!r}. If 'STALE', the adopt "
f"fallback adopted a stale final-answer event again."
)
# The instance attribute keeps the announced turn id; the adopted
# path is only the local rebinding inside ``_event_turn_matches``.
assert session.active_turn_id is None
async def test_run_turn_still_adopts_non_terminal_first_event() -> None:
"""Non-terminal first events (deltas, tool calls) must still be
adopted; the narrowing only refuses terminal final-answer items.
"""
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd="/tmp/workspace",
env={},
tool_executor=None,
)
session.start = AsyncMock()
session._proc = _FakeProcess()
session.thread_id = "thread-1"
session._request = AsyncMock(return_value={"result": {"turn": {"id": "turn-2"}}})
async def _inject_events() -> None:
await asyncio.sleep(0.01)
# First event: a non-terminal delta carrying a different id.
# Adopt should kick in (the original Codex quirk).
session._events.put_nowait(
{
"method": "item/agentMessage/delta",
"params": {
"turnId": "turn-2-rewritten",
"itemId": "msg-1",
"delta": "Hello",
},
}
)
# Final answer with the adopted id, terminating the turn.
session._events.put_nowait(
{
"method": "item/completed",
"params": {
"turnId": "turn-2-rewritten",
"item": {
"type": "agentMessage",
"id": "msg-1",
"text": "Hello",
"phase": "final_answer",
},
},
}
)
inject_task = asyncio.create_task(_inject_events())
events = [
event
async for event in session.run_turn(
messages=[{"role": "user", "content": "hi"}],
tools=[],
system_prompt="",
model="gpt-5.4-mini",
cwd=".",
sandbox="workspace-write",
)
]
await inject_task
turn_completes = [e for e in events if isinstance(e, TurnComplete)]
assert len(turn_completes) == 1
# The adopted id's deltas + final answer flow through normally.
assert turn_completes[0].response == "Hello"
def test_format_codex_error_params_extracts_provider_error_from_nested_error() -> None:
"""
Codex App Server's ``method == "error"`` frames carry the
actual provider failure inside ``params["error"]["message"]``,
typically as a stringified JSON blob like
``'{"error_code":"BAD_REQUEST","message":"..."}'``.
The formatter must unwrap two layers (the codex envelope
and the provider's JSON) so the user sees the human-readable
line, not the raw nested structure.
What breaks if this fails: users hit a config mismatch (Claude
model on a codex harness, bad profile, unsupported model on the
Databricks Responses passthrough) and see only the bare fallback
"Codex App Server error" — no clue why. The 2026-04-28 user
report was exactly this: codex+claude-opus-4-6 on Databricks
surfaced as "Codex App Server error" with zero diagnostic info.
"""
from omnigent.inner.codex_executor import _format_codex_error_params
params = {
"error": {
"message": (
'{"error_code":"BAD_REQUEST","message":'
'"Responses API passthrough is not supported for '
'model databricks-claude-opus-4-6."}'
),
"codexErrorInfo": "other",
"additionalDetails": None,
},
"willRetry": False,
"threadId": "abc",
"turnId": "def",
}
result = _format_codex_error_params(params)
# Provider's user-facing message must appear, with the error_code
# alongside it.
assert "Responses API passthrough is not supported" in result
assert "BAD_REQUEST" in result
# The bare fallback string must NOT appear — that's the "you
# silently ate my diagnostic" symptom this fix prevents.
assert "no params" not in result
# The codex-envelope's ``codexErrorInfo: "other"`` is generic
# and should be dropped from the output (we filter it
# explicitly to keep the message tight).
assert "codexErrorInfo" not in result
def test_format_codex_error_params_falls_back_to_raw_when_no_known_fields() -> None:
"""
A truly opaque error frame (no message / code / data / nested
error.message) must still surface SOMETHING — dump the raw
params dict. We never want the user to see a bare
"Codex App Server error" with no hint about what went wrong.
"""
from omnigent.inner.codex_executor import _format_codex_error_params
params = {"someField": "someValue", "willRetry": False}
result = _format_codex_error_params(params)
assert "raw_params" in result
assert "someField" in result
def test_format_codex_error_params_handles_missing_params() -> None:
"""
None / empty / non-dict params must produce a stable fallback
string — never crash, never empty.
"""
from omnigent.inner.codex_executor import _format_codex_error_params
assert "no params" in _format_codex_error_params(None)
assert "no params" in _format_codex_error_params({})
assert "no params" in _format_codex_error_params("not a dict")
def test_extract_codex_last_turn_usage_splits_cached_out_of_input() -> None:
"""``tokenUsage.last`` maps onto TurnComplete.usage, splitting cached tokens.
Codex's ``inputTokens`` is inclusive of cached tokens, so the cached
portion is moved into ``cache_read_input_tokens`` and ``input_tokens``
keeps only the non-cached remainder — otherwise the server prices cached
tokens at the full input rate. If ``input_tokens`` came back as 7 (not 6)
and there were no ``cache_read_input_tokens`` key, the split regressed.
"""
from omnigent.inner.codex_executor import _extract_codex_last_turn_usage
params = {
"threadId": "t1",
"turnId": "turn-1",
"tokenUsage": {
"last": {
"inputTokens": 7,
"outputTokens": 3,
"totalTokens": 10,
"cachedInputTokens": 1,
"reasoningOutputTokens": 0,
},
"total": {"inputTokens": 100, "outputTokens": 50, "totalTokens": 150},
},
}
assert _extract_codex_last_turn_usage(params) == {
"input_tokens": 6, # 7 total input - 1 cached
"output_tokens": 3,
"total_tokens": 10,
"cache_read_input_tokens": 1,
}
def test_extract_codex_last_turn_usage_no_cache_key_when_uncached() -> None:
"""No ``cachedInputTokens`` ⇒ input unchanged and no cache_read key added.
Guards against synthesizing a phantom cache bucket (which would shrink
the non-cached input the server bills at the full rate).
"""
from omnigent.inner.codex_executor import _extract_codex_last_turn_usage
params = {"tokenUsage": {"last": {"inputTokens": 7, "outputTokens": 3, "totalTokens": 10}}}
assert _extract_codex_last_turn_usage(params) == {
"input_tokens": 7,
"output_tokens": 3,
"total_tokens": 10,
}
def test_extract_codex_last_turn_usage_handles_missing_or_malformed() -> None:
"""Missing or non-dict shapes return None rather than raising."""
from omnigent.inner.codex_executor import _extract_codex_last_turn_usage
assert _extract_codex_last_turn_usage(None) is None
assert _extract_codex_last_turn_usage("not a dict") is None
assert _extract_codex_last_turn_usage({}) is None
assert _extract_codex_last_turn_usage({"tokenUsage": None}) is None
assert _extract_codex_last_turn_usage({"tokenUsage": {"total": {}}}) is None
def _make_skill_dir(root: Path, name: str) -> Path:
"""Create a minimal valid skill directory for the populator tests."""
skill_dir = root / name
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text(f"---\nname: {name}\n---\n# {name}\n")
return skill_dir
def test_populate_codex_skills_all(tmp_path: Path) -> None:
"""``skills_filter='all'`` symlinks every available skill from
every source.
Mirrors the SDK semantics where ``skills='all'`` exposes
every host-discovered skill. The populator is the codex
equivalent — it materializes the union of all sources into
the per-conversation ``$CODEX_HOME/skills/`` so codex's
auto-discovery surfaces them all.
"""
from omnigent.inner.codex_executor import _populate_codex_skills
host_skills = tmp_path / "host"
bundle_skills = tmp_path / "bundle"
target = tmp_path / "codex_home_skills"
_make_skill_dir(host_skills, "alpha")
_make_skill_dir(host_skills, "beta")
_make_skill_dir(bundle_skills, "gamma")
_populate_codex_skills(target, "all", [bundle_skills, host_skills])
# Every skill from both sources is present as a symlink. If
# the populator only scanned one source, this would miss
# either ``gamma`` (bundle-only) or ``alpha``/``beta``
# (host-only) — the failure message would name the missing
# entry.
assert sorted(p.name for p in target.iterdir()) == ["alpha", "beta", "gamma"]
assert all((target / name).is_symlink() for name in ["alpha", "beta", "gamma"])
def test_populate_codex_skills_none(tmp_path: Path) -> None:
"""``skills_filter='none'`` leaves the target dir absent
entirely.
Codex's discovery walks ``$CODEX_HOME/skills/`` — if the
directory doesn't exist, no skills load. This is the
hermetic-agent regression-pin: Omnigent must produce no skill
surface for ``skills: none`` even when host
``~/.codex/skills/`` is populated.
"""
from omnigent.inner.codex_executor import _populate_codex_skills
host_skills = tmp_path / "host"
target = tmp_path / "codex_home_skills"
_make_skill_dir(host_skills, "alpha")
_populate_codex_skills(target, "none", [host_skills])
# Target dir must not exist (codex's scan no-ops cleanly when
# the path is absent). A directory with zero entries would
# still pass codex's existence check but is wasteful — the
# absent-dir contract is tighter.
assert not target.exists()
def test_populate_codex_skills_named_subset(tmp_path: Path) -> None:
"""``skills_filter=[name, ...]`` exposes only the named
skills.
Names not present in any source are silently skipped (the
SDK semantic — missing skill is no-op, not an error).
Names present in multiple sources resolve to the
first-listed source (so callers can express priority by
ordering ``sources``).
"""
from omnigent.inner.codex_executor import _populate_codex_skills
host_skills = tmp_path / "host"
bundle_skills = tmp_path / "bundle"
target = tmp_path / "codex_home_skills"
_make_skill_dir(host_skills, "alpha")
_make_skill_dir(host_skills, "beta")
_make_skill_dir(bundle_skills, "alpha") # name collision
_make_skill_dir(bundle_skills, "gamma")
_populate_codex_skills(
target,
["alpha", "gamma", "missing_skill"],
[bundle_skills, host_skills], # bundle wins
)
# ``alpha`` resolves to the bundle copy (first source listed).
# ``gamma`` is bundle-only. ``beta`` is host-only and not
# named — must NOT appear. ``missing_skill`` doesn't exist
# anywhere — must be silently dropped, not raised.
assert sorted(p.name for p in target.iterdir()) == ["alpha", "gamma"]
alpha_target = (target / "alpha").resolve()
assert alpha_target == (bundle_skills / "alpha").resolve(), (
f"name collision should resolve to first source (bundle), but resolved to {alpha_target}"
)
def test_populate_codex_skills_from_bundle_links_bundle_skills(tmp_path: Path) -> None:
"""
``populate_codex_skills_from_bundle`` links a bundle's ``skills/`` into
``<codex_home>/skills/``.
This is the shared helper both the wrapped codex executor and the
codex-native launch path use, so a deployed agent's bundled skills
reach the native Codex CLI (the gap before this change: codex-native
populated no skills). Fails if the bundle source isn't scanned or the
target isn't created under ``<codex_home>/skills``.
"""
from omnigent.inner.codex_executor import populate_codex_skills_from_bundle
bundle = tmp_path / "bundle"
_make_skill_dir(bundle / "skills", "authoring")
codex_home = tmp_path / "codex_home"
populate_codex_skills_from_bundle(codex_home, bundle, "all")
linked = codex_home / "skills" / "authoring"
assert linked.is_symlink() or linked.is_dir()
assert (linked / "SKILL.md").is_file()
def test_populate_codex_skills_from_bundle_none_leaves_no_dir(tmp_path: Path) -> None:
"""
``skills_filter="none"`` produces no ``skills/`` dir even when the
bundle ships skills — the codex-native parity for a hermetic agent.
"""
from omnigent.inner.codex_executor import populate_codex_skills_from_bundle
bundle = tmp_path / "bundle"
_make_skill_dir(bundle / "skills", "authoring")
codex_home = tmp_path / "codex_home"
populate_codex_skills_from_bundle(codex_home, bundle, "none")
assert not (codex_home / "skills").exists()
# ---------------------------------------------------------------------------
# _populate_codex_home_config tests
# ---------------------------------------------------------------------------
def test_populate_codex_home_config_symlinks_auth_and_config(tmp_path: Path) -> None:
"""``auth.json`` is symlinked; ``config.toml`` is copied (not symlinked).
``auth.json`` is symlinked so OAuth token refreshes written to the real
home propagate to running sessions. ``config.toml`` is copied so an
in-TUI ``/model`` command writes only to the session's private copy and
never mutates the shared ``~/.codex/config.toml``.
"""
from omnigent.inner.codex_executor import _populate_codex_home_config
source = tmp_path / "real_codex_home"
source.mkdir()
(source / "auth.json").write_text('{"auth_mode": "chatgpt"}')
(source / "config.toml").write_text('[default]\nmodel = "gpt-5.4"')
target = tmp_path / "temp_codex_home"
target.mkdir()
_populate_codex_home_config(target, source)
# auth.json: symlink so live credential refreshes propagate.
assert (target / "auth.json").is_symlink()
assert (target / "auth.json").read_text() == '{"auth_mode": "chatgpt"}'
# config.toml: independent copy so /model writes stay session-local.
assert not (target / "config.toml").is_symlink()
assert (target / "config.toml").is_file()
assert (target / "config.toml").read_text() == '[default]\nmodel = "gpt-5.4"'
def test_populate_codex_home_config_config_toml_copy_is_isolated(tmp_path: Path) -> None:
"""Writing to the session's ``config.toml`` copy does not affect the source.
Proves that ``/model`` inside a session cannot mutate the shared
``~/.codex/config.toml`` and silently change another session's model or
cost-policy enforcement.
"""
from omnigent.inner.codex_executor import _populate_codex_home_config
source = tmp_path / "real_codex_home"
source.mkdir()
(source / "config.toml").write_text('[default]\nmodel = "gpt-5.4"')
target = tmp_path / "temp_codex_home"
target.mkdir()
_populate_codex_home_config(target, source)
# Simulate a /model write inside the session.
(target / "config.toml").write_text('[default]\nmodel = "gpt-4o-mini"')
# Source must be untouched.
assert (source / "config.toml").read_text() == '[default]\nmodel = "gpt-5.4"'
def test_populate_codex_home_config_missing_source_dir(tmp_path: Path) -> None:
"""When the source ``CODEX_HOME`` dir doesn't exist (fresh install),
nothing is created in the target.
Handles the case where codex has never been run locally — the
populator must no-op cleanly rather than raising on a missing path.
"""
from omnigent.inner.codex_executor import _populate_codex_home_config
source = tmp_path / "nonexistent"
target = tmp_path / "temp_codex_home"
target.mkdir()
_populate_codex_home_config(target, source)
assert list(target.iterdir()) == []
def test_populate_codex_home_config_partial_files(tmp_path: Path) -> None:
"""When only some config files exist, only those are symlinked.
API-key users may not have ``auth.json`` (subscription auth is
opt-in via ``codex auth``); ``config.toml`` is optional too. The
populator must skip missing files silently.
"""
from omnigent.inner.codex_executor import _populate_codex_home_config
source = tmp_path / "real_codex_home"
source.mkdir()
(source / "auth.json").write_text('{"auth_mode": "chatgpt"}')
# config.toml intentionally absent
target = tmp_path / "temp_codex_home"
target.mkdir()
_populate_codex_home_config(target, source)
assert (target / "auth.json").is_symlink()
assert not (target / "config.toml").exists()
def test_app_server_start_uses_real_home_for_private_inherited_codex_home(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""
Empty inherited ``CODEX_HOME`` does not hide the real user login at startup.
Omnigent itself can be launched from a Codex-managed environment
where ``CODEX_HOME`` points at an isolated private home. If that
inherited home lacks Codex auth/config files, app-server startup must
bridge from the user's real ``~/.codex`` equivalent so Codex sessions
do not prompt to log in despite the regular CLI already being authenticated.
:param tmp_path: Temporary directory for isolated Codex homes.
:param monkeypatch: Pytest fixture used to isolate ``HOME`` and
inherited ``CODEX_HOME``.
:returns: None.
"""
home = tmp_path / "home"
monkeypatch.setenv("HOME", str(home))
inherited = home / ".omnigent" / "codex-native" / "abc123" / "codex-home"
inherited.mkdir(parents=True)
monkeypatch.setenv("CODEX_HOME", str(inherited))
real_codex_home = home / ".codex"
real_codex_home.mkdir()
(real_codex_home / "auth.json").write_text('{"auth_mode": "api_key"}')
(real_codex_home / "config.toml").write_text('model_provider = "openai"')
async def _t() -> None:
"""
Drive ``_CodexAppServerSession.start`` with the subprocess boundary stubbed.
:returns: None.
"""
fake_proc = _FakeProcess()
recorded_env: dict[str, str] | None = None
target_at_spawn: Path | None = None
async def _fake_create_subprocess_exec(*args, **kwargs):
"""
Capture the app-server subprocess environment.
:param args: Positional subprocess argv values.
:param kwargs: Keyword subprocess options, including ``env``.
:returns: Fake subprocess handle.
"""
nonlocal recorded_env, target_at_spawn
recorded_env = kwargs.get("env")
assert recorded_env is not None
target_at_spawn = Path(recorded_env["CODEX_HOME"])
assert target_at_spawn != inherited
assert (target_at_spawn / "auth.json").is_symlink()
assert not (target_at_spawn / "config.toml").is_symlink()
assert (target_at_spawn / "config.toml").is_file()
assert (target_at_spawn / "auth.json").read_text() == '{"auth_mode": "api_key"}'
assert (target_at_spawn / "config.toml").read_text() == 'model_provider = "openai"'
return fake_proc
workspace = tmp_path / "workspace"
workspace.mkdir()
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd=str(workspace),
env={},
tool_executor=None,
)
session._request = AsyncMock(return_value={"result": {}})
with patch(
"omnigent.inner.codex_executor._create_subprocess_exec",
new=_fake_create_subprocess_exec,
):
await session.start()
assert recorded_env is not None
assert target_at_spawn is not None
await session.close()
assert fake_proc.terminated
_run(_t())
def test_app_server_start_preserves_custom_home_from_inherited_private_symlink(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""
Nested startup preserves a parent's custom Codex home source.
A top-level launch may bridge auth/config from an explicit custom
``CODEX_HOME`` into an Omnigent private home. A nested launch inherits
only that private path, so it must infer the original custom source from
the existing symlink targets instead of falling back to ``~/.codex``.
:param tmp_path: Temporary directory for isolated Codex homes.
:param monkeypatch: Pytest fixture used to isolate ``HOME`` and
inherited ``CODEX_HOME``.
:returns: None.
"""
home = tmp_path / "home"
monkeypatch.setenv("HOME", str(home))
custom_home = tmp_path / "custom-codex-home"
custom_home.mkdir()
(custom_home / "auth.json").write_text('{"auth_mode": "custom"}')
(custom_home / "config.toml").write_text('model_provider = "custom"')
default_home = home / ".codex"
default_home.mkdir(parents=True)
(default_home / "auth.json").write_text('{"auth_mode": "default"}')
(default_home / "config.toml").write_text('model_provider = "default"')
inherited = home / ".omnigent" / "codex-native" / "abc123" / "codex-home"
inherited.mkdir(parents=True)
(inherited / "auth.json").symlink_to(custom_home / "auth.json")
(inherited / "config.toml").symlink_to(custom_home / "config.toml")
monkeypatch.setenv("CODEX_HOME", str(inherited))
async def _t() -> None:
"""
Drive ``_CodexAppServerSession.start`` with a stubbed subprocess.
:returns: None.
"""
fake_proc = _FakeProcess()
async def _fake_create_subprocess_exec(*args, **kwargs):
"""
Assert the bridged files come from the custom home at spawn time.
:param args: Positional subprocess argv values.
:param kwargs: Keyword subprocess options, including ``env``.
:returns: Fake subprocess handle.
"""
target = Path(kwargs["env"]["CODEX_HOME"])
assert (target / "auth.json").read_text() == '{"auth_mode": "custom"}'
assert (target / "config.toml").read_text() == 'model_provider = "custom"'
return fake_proc
workspace = tmp_path / "workspace"
workspace.mkdir()
session = _CodexAppServerSession(
codex_path="/bin/echo",
cwd=str(workspace),
env={},
tool_executor=None,
)
session._request = AsyncMock(return_value={"result": {}})
with patch(
"omnigent.inner.codex_executor._create_subprocess_exec",
new=_fake_create_subprocess_exec,
):
await session.start()
await session.close()
_run(_t())
def test_populate_codex_home_config_does_not_overwrite_existing(tmp_path: Path) -> None:
"""If a config file already exists in the target (e.g. from a
previous partial start), it is not replaced.
Guards against double-start races or manual overrides placed
in the temp dir before the populator runs.
"""
from omnigent.inner.codex_executor import _populate_codex_home_config
source = tmp_path / "real_codex_home"
source.mkdir()
(source / "auth.json").write_text('{"new": true}')
target = tmp_path / "temp_codex_home"
target.mkdir()
(target / "auth.json").write_text('{"old": true}')
_populate_codex_home_config(target, source)
assert (target / "auth.json").read_text() == '{"old": true}'
assert not (target / "auth.json").is_symlink()
# ---------------------------------------------------------------------------
# _clean_codex_env tests
# ---------------------------------------------------------------------------
def test_clean_codex_env_excludes_openai_api_key(monkeypatch) -> None:
"""``_clean_codex_env`` must strip ``OPENAI_API_KEY`` even though
the ``OPENAI_`` prefix is in the allowlist.
The codex harness routes through the CLI's subscription auth
(``auth.json``), not a developer API key. Leaking
``OPENAI_API_KEY`` into the subprocess would charge the user's
developer account instead of their subscription plan.
"""
from omnigent.inner.codex_executor import _clean_codex_env
monkeypatch.setenv("OPENAI_API_KEY", "sk-test-secret")
monkeypatch.setenv("OPENAI_MAX_RETRIES", "3")
monkeypatch.setenv("OPENAI_TIMEOUT", "60")
env = _clean_codex_env()
assert "OPENAI_API_KEY" not in env, (
"OPENAI_API_KEY must be excluded — subscription auth, not API key"
)
# Other OPENAI_* vars (retry/timeout knobs) must still pass through.
assert env.get("OPENAI_MAX_RETRIES") == "3"
assert env.get("OPENAI_TIMEOUT") == "60"
def test_clean_codex_env_includes_databricks_bearer(monkeypatch) -> None:
"""``_clean_codex_env`` preserves CI's explicit Databricks bearer.
:param monkeypatch: Pytest monkeypatch fixture.
"""
from omnigent.inner.codex_executor import _clean_codex_env
monkeypatch.setenv("DATABRICKS_BEARER", "ci-bearer")
monkeypatch.setenv("DATABRICKS_TOKEN", "stale-token")
env = _clean_codex_env()
assert env.get("DATABRICKS_BEARER") == "ci-bearer"
assert "DATABRICKS_TOKEN" not in env
def test_clean_codex_env_includes_omnigent_session_marker(monkeypatch) -> None:
"""The ``OMNIGENT`` session marker survives the codex env scrub.
The marker (set once on the runner) must reach the codex CLI so the
shell commands codex runs can detect they are inside an Omnigent
session, like ``CLAUDE_CODE`` / ``CODEX``.
:param monkeypatch: Pytest monkeypatch fixture.
"""
from omnigent.inner.codex_executor import _clean_codex_env
from omnigent.runner.identity import (
OMNIGENT_SESSION_ENV_VALUE,
OMNIGENT_SESSION_ENV_VAR,
)
monkeypatch.setenv(OMNIGENT_SESSION_ENV_VAR, OMNIGENT_SESSION_ENV_VALUE)
env = _clean_codex_env()
assert env.get(OMNIGENT_SESSION_ENV_VAR) == OMNIGENT_SESSION_ENV_VALUE
# ---------------------------------------------------------------------------
# Tests for _to_codex_input_items — input_file → inline text conversion
# ---------------------------------------------------------------------------
def test_to_codex_input_items_text_block_passes_through() -> None:
"""Plain text blocks are mapped to Codex ``{"type": "text"}`` items."""
result = _to_codex_input_items([{"type": "input_text", "text": "hello"}])
# One item produced, with the same text — nothing stripped or altered.
assert result == [{"type": "text", "text": "hello"}]
def test_to_codex_input_items_image_block_passes_through() -> None:
"""Image blocks are mapped to Codex ``{"type": "image"}`` items."""
result = _to_codex_input_items(
[{"type": "input_image", "image_url": "https://example.com/img.png"}]
)
assert result == [{"type": "image", "url": "https://example.com/img.png"}]
def test_to_codex_input_items_input_file_data_uri_decoded_to_text() -> None:
"""``input_file`` with a ``data:`` URI is decoded and emitted as text.
This is the primary new behaviour: the Codex CLI does not support
``input_file`` blocks, so the base64 payload must be decoded inline.
"""
content = "# Hello\nThis is a markdown file."
b64 = base64.b64encode(content.encode()).decode()
block = {"type": "input_file", "file_data": f"data:text/plain;base64,{b64}"}
result = _to_codex_input_items([block])
# Decoded text emitted as a plain ``text`` item.
assert result == [{"type": "text", "text": content}], (
f"Expected decoded file content as text item, got: {result!r}"
)
def test_to_codex_input_items_input_file_plain_string_used_as_text() -> None:
"""``input_file`` whose ``file_data`` is NOT a data URI is used as-is."""
block = {"type": "input_file", "file_data": "plain text content"}
result = _to_codex_input_items([block])
# When file_data is already plain text (no "data:" prefix), use it directly.
assert result == [{"type": "text", "text": "plain text content"}]
def test_to_codex_input_items_input_file_empty_file_data_dropped() -> None:
"""``input_file`` with empty ``file_data`` produces no output item.
An empty block would add noise without content; dropping it is correct.
"""
result = _to_codex_input_items([{"type": "input_file", "file_data": ""}])
# No item emitted — empty content is silently dropped.
assert result == [], f"Expected empty list, got: {result!r}"
def test_to_codex_input_items_input_file_malformed_base64_dropped() -> None:
"""``input_file`` with an invalid base64 payload produces no output item.
The decode path catches all exceptions and falls back to empty string,
which is then dropped. The caller should not see an error.
"""
block = {"type": "input_file", "file_data": "data:text/plain;base64,!!!not-valid-base64!!!"}
result = _to_codex_input_items([block])
# Malformed base64 → decode fails → text="" → dropped.
assert result == [], f"Expected empty list, got: {result!r}"
def test_to_codex_input_items_binary_file_dropped() -> None:
"""``input_file`` with a binary (non-text) MIME type is silently dropped.
Decoding a PDF as UTF-8 produces garbled replacement characters that waste
tokens and provide no useful information to the model.
"""
b64 = base64.b64encode(b"%PDF-1.4 binary content").decode()
block = {"type": "input_file", "file_data": f"data:application/pdf;base64,{b64}"}
result = _to_codex_input_items([block])
# Binary MIME type → dropped, not inlined as garbage text.
assert result == [], f"Expected binary block to be dropped, got: {result!r}"
def test_to_codex_input_items_mixed_blocks() -> None:
"""Text + image + input_file blocks are all handled in one pass."""
file_content = "file body"
b64 = base64.b64encode(file_content.encode()).decode()
blocks = [
{"type": "input_text", "text": "question"},
{"type": "input_file", "file_data": f"data:text/plain;base64,{b64}"},
{"type": "input_image", "image_url": "https://example.com/pic.jpg"},
]
result = _to_codex_input_items(blocks)
assert result == [
{"type": "text", "text": "question"},
{"type": "text", "text": "file body"},
{"type": "image", "url": "https://example.com/pic.jpg"},
], f"Unexpected result: {result!r}"
@dataclass
class _FakeVersionProcess:
"""
Minimal subprocess stub for ``codex --version`` parsing tests.
:param stdout: Bytes the stubbed process emits on stdout, e.g.
``b"codex-cli 0.129.0\\n"``.
"""
stdout: bytes
async def communicate(self) -> tuple[bytes, bytes]:
"""
Return the canned ``(stdout, stderr)`` pair.
:returns: ``(self.stdout, b"")`` — the version probe ignores
stderr.
"""
return (self.stdout, b"")
@pytest.mark.parametrize(
"output,expected",
[
(b"codex-cli 0.129.0\n", (0, 129, 0)),
(b"codex-cli 0.136.0\n", (0, 136, 0)),
# Pre-release suffix: only the X.Y.Z core is parsed, so an alpha
# of a supported release compares as that release.
(b"codex-cli 0.132.0-alpha.1\n", (0, 132, 0)),
# No X.Y.Z token → unknown (caller treats None as "supported").
(b"codex-cli (unknown build)\n", None),
],
)
async def test_codex_cli_version_parses_output(
monkeypatch: pytest.MonkeyPatch,
output: bytes,
expected: tuple[int, int, int] | None,
) -> None:
"""
``_codex_cli_version`` parses the numeric core of ``codex --version``.
Guards the version gate that decides whether the native policy hook
can be registered (>= 0.129). A parsing regression here would either
disable enforcement on a supported codex or register an un-trustable
hook on an old one.
"""
async def _fake_exec(*_args: Any, **_kwargs: Any) -> _FakeVersionProcess:
return _FakeVersionProcess(stdout=output)
monkeypatch.setattr("omnigent.inner.codex_executor._create_subprocess_exec", _fake_exec)
assert await _codex_cli_version("/usr/local/bin/codex") == expected
async def test_codex_cli_version_returns_none_on_oserror(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
A codex binary that cannot be executed yields ``None``, not a crash.
``None`` means "version unknown" and the caller proceeds (treats it
as supported). Fails if an OSError from a missing/broken binary
propagates out of the probe.
"""
async def _boom(*_args: Any, **_kwargs: Any) -> None:
raise OSError("no such binary")
monkeypatch.setattr("omnigent.inner.codex_executor._create_subprocess_exec", _boom)
assert await _codex_cli_version("/usr/local/bin/codex") is None
async def test_codex_cli_version_times_out_and_kills_proc(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""
A hung ``codex --version`` is killed and reported as unknown.
Guards session startup: without the probe timeout, a codex build that
blocks on ``--version`` would stall ``start()`` indefinitely. The probe
must return ``None`` (treated as "supported, proceed") and kill the
stuck process rather than leak it.
"""
class _HangingProcess:
"""Stub process whose ``communicate`` blocks until killed."""
def __init__(self) -> None:
self.killed = False
async def communicate(self) -> tuple[bytes, bytes]:
# Block far longer than the probe timeout so wait_for fires.
await asyncio.sleep(60)
return (b"", b"")
def kill(self) -> None:
self.killed = True
async def wait(self) -> int:
return -9
proc = _HangingProcess()
async def _fake_exec(*_args: Any, **_kwargs: Any) -> _HangingProcess:
return proc
# Shrink the probe budget so the test does not actually wait the full 5s.
monkeypatch.setattr("omnigent.inner.codex_executor._CODEX_VERSION_PROBE_TIMEOUT_SECONDS", 0.05)
monkeypatch.setattr("omnigent.inner.codex_executor._create_subprocess_exec", _fake_exec)
assert await _codex_cli_version("/usr/local/bin/codex") is None
# The stuck process was killed, not leaked.
assert proc.killed is True
# ── model_provider_override (cli-config / subscription pinning) ─────────────
# Function-based (the project standard); the TestCase class above predates it.
def test_model_provider_override_appends_pin() -> None:
"""The override becomes exactly one ``-c model_provider=...`` fragment.
json.dumps quoting yields a valid TOML basic string. Failure means an
adopted cli-config provider (or the subscription's openai pin) never
reaches the codex CLI, so the bridged config.toml's default provider
silently routes the session instead.
"""
with patch("omnigent.inner.codex_executor._find_codex_cli", return_value="/usr/bin/codex"):
executor = CodexExecutor(model_provider_override="Databricks")
assert executor._codex_config_overrides == ['model_provider="Databricks"']
def test_model_provider_override_with_gateway_raises() -> None:
"""gateway=True and model_provider_override are mutually exclusive.
Both write ``model_provider`` into the -c overrides; last-one-wins
would silently drop the gateway transport. Failure (no exception)
means a misconfigured AP producer ships an ambiguous launch.
"""
with (
patch("omnigent.inner.codex_executor._find_codex_cli", return_value="/usr/bin/codex"),
pytest.raises(OSError, match="mutually exclusive"),
):
CodexExecutor(
gateway=True,
gateway_host="https://gw.example.com",
base_url_override="https://gw.example.com/codex/v1",
gateway_auth_command="printf %s tok",
model="some-model",
model_provider_override="Databricks",
)
def _mk_codex_skill(skills_dir: Path, name: str) -> None:
"""Create a ``<skills_dir>/<name>/SKILL.md`` skill directory."""
d = skills_dir / name
d.mkdir(parents=True)
(d / "SKILL.md").write_text(f"---\nname: {name}\ndescription: d\n---\nbody\n")
def test_select_codex_skill_dirs_all_first_source_wins(tmp_path: Path) -> None:
from omnigent.inner.codex_executor import select_codex_skill_dirs
a, b = tmp_path / "a", tmp_path / "b"
_mk_codex_skill(a, "shared")
_mk_codex_skill(b, "shared")
_mk_codex_skill(b, "only-b")
out = select_codex_skill_dirs("all", [a, b])
assert out["shared"] == a / "shared"
assert out["only-b"] == b / "only-b"
def test_select_codex_skill_dirs_none_and_list(tmp_path: Path) -> None:
from omnigent.inner.codex_executor import select_codex_skill_dirs
a = tmp_path / "a"
_mk_codex_skill(a, "x")
_mk_codex_skill(a, "y")
assert select_codex_skill_dirs("none", [a]) == {}
assert set(select_codex_skill_dirs(["x"], [a])) == {"x"}
def test_codex_skill_sources_order_bundle_then_host(tmp_path: Path) -> None:
"""codex_skill_sources lists <bundle>/skills before <home>/.codex/skills."""
from omnigent.inner.codex_executor import codex_skill_sources
bundle = tmp_path / "bundle"
(bundle / "skills").mkdir(parents=True)
home = tmp_path / "home"
(home / ".codex" / "skills").mkdir(parents=True)
assert codex_skill_sources(bundle, home) == [
bundle / "skills",
home / ".codex" / "skills",
]
def test_codex_skill_sources_omits_absent_dirs(tmp_path: Path) -> None:
"""Only existing dirs are returned (bundle absent → host only)."""
from omnigent.inner.codex_executor import codex_skill_sources
home = tmp_path / "home"
(home / ".codex" / "skills").mkdir(parents=True)
assert codex_skill_sources(None, home) == [home / ".codex" / "skills"]
assert codex_skill_sources(tmp_path / "no-bundle", home) == [home / ".codex" / "skills"]
def test_clean_codex_env_honors_extra_allow(monkeypatch):
from omnigent.inner.codex_executor import _clean_codex_env
monkeypatch.setenv("CRAWL4AI_API_TOKEN", "secret-tok")
monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "ch-key")
# undeclared → stripped by the hardcoded allowlist
assert "CRAWL4AI_API_TOKEN" not in _clean_codex_env()
# declared via env_passthrough → admitted
env = _clean_codex_env(["CRAWL4AI_API_TOKEN", "COMPANIES_HOUSE_API_KEY"])
assert env["CRAWL4AI_API_TOKEN"] == "secret-tok"
assert env["COMPANIES_HOUSE_API_KEY"] == "ch-key"
def test_clean_codex_env_deny_wins_over_extra_allow(monkeypatch):
from omnigent.inner.codex_executor import _clean_codex_env
monkeypatch.setenv("OPENAI_API_KEY", "sk-stripped")
# the deny rule (subscription auth) wins even if a caller declares it
assert "OPENAI_API_KEY" not in _clean_codex_env(["OPENAI_API_KEY"])
def test_declared_passthrough_reads_sandbox_env_passthrough():
from omnigent.inner.codex_executor import _declared_passthrough
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
# env_passthrough lives on os_env.sandbox, not os_env directly
spec = OSEnvSpec(
sandbox=OSEnvSandboxSpec(
type="none", env_passthrough=["CRAWL4AI_API_TOKEN", "COMPANIES_HOUSE_API_KEY"]
)
)
assert _declared_passthrough(spec) == ("CRAWL4AI_API_TOKEN", "COMPANIES_HOUSE_API_KEY")
# guards: None os_env / None sandbox / unset list all yield ()
assert _declared_passthrough(None) == ()
assert _declared_passthrough(OSEnvSpec(sandbox=None)) == ()
assert _declared_passthrough(OSEnvSpec(sandbox=OSEnvSandboxSpec(type="none"))) == ()
Reference in New Issue View Git Blame Copy Permalink