diff --git a/studio/backend/core/inference/codex_spoof.py b/studio/backend/core/inference/codex_spoof.py index 885041b4ff..ea18133dc7 100644 --- a/studio/backend/core/inference/codex_spoof.py +++ b/studio/backend/core/inference/codex_spoof.py @@ -61,7 +61,7 @@ class SandboxMode(str, Enum): class AppServerConfig: env: Optional[dict[str, str]] = None codex_bin: Optional[str] = None - extra: dict[str, Any] = field(default_factory=dict) + extra: dict[str, Any] = field(default_factory = dict) @dataclass @@ -97,7 +97,7 @@ def _tab_id_from_system(system: Optional[str]) -> int: line = line.strip() if line.startswith("[tab ") and line.endswith("]"): try: - return int(line[len("[tab "): -1].split("/")[0]) + return int(line[len("[tab ") : -1].split("/")[0]) except ValueError: pass return 0 @@ -152,7 +152,9 @@ class _TurnStream: # Final completion event -- the canonical SDK always emits this, # and ``_stream_thread_run`` uses it as its fallback when no # deltas arrived (so worth keeping even when deltas did stream). - yield _ItemCompletedNotification(item=_ItemRoot(root=_AgentMessage(text=self._text))) + yield _ItemCompletedNotification( + item = _ItemRoot(root = _AgentMessage(text = self._text)) + ) async def __anext__(self) -> Any: if self._iter is None: @@ -198,7 +200,7 @@ class _Thread: text = _spoof_response_text(self._model, prompt, self._tab_id) await asyncio.sleep(0) - return SimpleNamespace(text=text, final_response=text) + return SimpleNamespace(text = text, final_response = text) class AsyncCodex: @@ -227,7 +229,7 @@ class AsyncCodex: # Either kwarg path is accepted -- the real provider tries # ``base_instructions`` first then falls back to ``system``. sys_text = base_instructions if base_instructions is not None else system - return _Thread(model=model, system=sys_text) + return _Thread(model = model, system = sys_text) def install_as_openai_codex() -> None: @@ -260,7 +262,7 @@ def _build_module_alias(name: str) -> Any: # ``importlib.util.find_spec(name)`` walks ``sys.modules[name].__spec__`` # first, so an empty spec is required for the provider's existing # availability probe to recognise the spoof. - mod.__spec__ = importlib.machinery.ModuleSpec(name, loader=None) + mod.__spec__ = importlib.machinery.ModuleSpec(name, loader = None) mod.AsyncCodex = AsyncCodex # type: ignore[attr-defined] mod.AppServerConfig = AppServerConfig # type: ignore[attr-defined] mod.ApprovalMode = ApprovalMode # type: ignore[attr-defined] diff --git a/studio/backend/tests/test_codex_provider.py b/studio/backend/tests/test_codex_provider.py index 77ff2b3fb3..a14f6faf56 100644 --- a/studio/backend/tests/test_codex_provider.py +++ b/studio/backend/tests/test_codex_provider.py @@ -2367,15 +2367,18 @@ class TestCodexSpoofModule: def test_install_swaps_in_module_when_flag_set(self, monkeypatch): # Ensure clean import state. import sys + for name in ("openai_codex", "codex_app_server"): sys.modules.pop(name, None) from core.inference import codex_spoof + monkeypatch.setenv(codex_spoof.SPOOF_ENV_VAR, "1") assert codex_spoof.is_spoof_enabled() codex_spoof.install_as_openai_codex() import openai_codex # type: ignore[import-not-found] + assert getattr(openai_codex, "__spoof__", False) is True assert hasattr(openai_codex, "AsyncCodex") assert hasattr(openai_codex, "AppServerConfig") @@ -2384,10 +2387,12 @@ class TestCodexSpoofModule: def test_flag_disabled_does_not_install(self, monkeypatch): import sys + for name in ("openai_codex", "codex_app_server"): sys.modules.pop(name, None) from core.inference import codex_spoof - monkeypatch.delenv(codex_spoof.SPOOF_ENV_VAR, raising=False) + + monkeypatch.delenv(codex_spoof.SPOOF_ENV_VAR, raising = False) assert not codex_spoof.is_spoof_enabled() def test_spoof_stream_emits_deltas_and_completion(self): @@ -2395,14 +2400,12 @@ class TestCodexSpoofModule: from core.inference import codex_spoof async def run(): - codex = codex_spoof.AsyncCodex( - config=codex_spoof.AppServerConfig(env={}) - ) + codex = codex_spoof.AsyncCodex(config = codex_spoof.AppServerConfig(env = {})) thread = await codex.thread_start( - model="gpt-5.4-mini", - base_instructions=None, - approval_mode=codex_spoof.ApprovalMode.deny_all, - sandbox=codex_spoof.SandboxMode.read_only, + model = "gpt-5.4-mini", + base_instructions = None, + approval_mode = codex_spoof.ApprovalMode.deny_all, + sandbox = codex_spoof.SandboxMode.read_only, ) events = [] async for ev in thread.turn("hello").stream(): @@ -2410,7 +2413,11 @@ class TestCodexSpoofModule: return events events = asyncio.run(run()) - deltas = [e for e in events if isinstance(e, dict) and e.get("type") == "message.delta"] + deltas = [ + e + for e in events + if isinstance(e, dict) and e.get("type") == "message.delta" + ] assert len(deltas) >= 3, "spoof should stream multiple deltas" last = events[-1] assert type(last).__name__ == "_ItemCompletedNotification" @@ -2426,8 +2433,8 @@ class TestCodexSpoofModule: async def reply_for_tab(idx: int) -> str: codex = codex_spoof.AsyncCodex() thread = await codex.thread_start( - model="gpt-5.4-mini", - base_instructions=f"[tab {idx}/3]", + model = "gpt-5.4-mini", + base_instructions = f"[tab {idx}/3]", ) result = await thread.run("explain LoRA") return result.text @@ -2442,23 +2449,27 @@ class TestCodexSpoofModule: tab_replies = asyncio.run(_gather()) # Each tab must mention its own worker index, so the UI tabs # show visibly distinct text when clicked. - for i, reply in enumerate(tab_replies, start=1): + for i, reply in enumerate(tab_replies, start = 1): assert f"worker {i}" in reply, f"tab {i} missing its tag: {reply}" def test_provider_picks_up_spoof_via_import(self, monkeypatch): import sys + for name in ("openai_codex", "codex_app_server"): sys.modules.pop(name, None) from core.inference import codex_provider, codex_spoof + monkeypatch.setenv(codex_spoof.SPOOF_ENV_VAR, "1") mod = codex_provider._import_codex() assert getattr(mod, "__spoof__", False) is True def test_safety_kwargs_resolve_against_spoof(self, monkeypatch): import sys + for name in ("openai_codex", "codex_app_server"): sys.modules.pop(name, None) from core.inference import codex_provider, codex_spoof + monkeypatch.setenv(codex_spoof.SPOOF_ENV_VAR, "1") codex_provider._import_codex() # ensures install kwargs = codex_provider._safe_thread_safety_kwargs()