1  import asyncio
  2  import sys
  3  import types
  4  from types import SimpleNamespace
  5  
  6  
  7  sys.modules.setdefault("fire", types.SimpleNamespace(Fire=lambda *a, **k: None))
  8  sys.modules.setdefault("firecrawl", types.SimpleNamespace(Firecrawl=object))
  9  sys.modules.setdefault("fal_client", types.SimpleNamespace())
 10  
 11  import cron.scheduler as cron_scheduler
 12  import gateway.run as gateway_run
 13  import run_agent
 14  from gateway.config import Platform
 15  from gateway.session import SessionSource
 16  
 17  
 18  def _patch_agent_bootstrap(monkeypatch):
 19      monkeypatch.setattr(
 20          run_agent,
 21          "get_tool_definitions",
 22          lambda **kwargs: [
 23              {
 24                  "type": "function",
 25                  "function": {
 26                      "name": "terminal",
 27                      "description": "Run shell commands.",
 28                      "parameters": {"type": "object", "properties": {}},
 29                  },
 30              }
 31          ],
 32      )
 33      monkeypatch.setattr(run_agent, "check_toolset_requirements", lambda: {})
 34  
 35  
 36  def _codex_message_response(text: str):
 37      return SimpleNamespace(
 38          output=[
 39              SimpleNamespace(
 40                  type="message",
 41                  content=[SimpleNamespace(type="output_text", text=text)],
 42              )
 43          ],
 44          usage=SimpleNamespace(input_tokens=5, output_tokens=3, total_tokens=8),
 45          status="completed",
 46          model="gpt-5-codex",
 47      )
 48  
 49  
 50  class _UnauthorizedError(RuntimeError):
 51      def __init__(self):
 52          super().__init__("Error code: 401 - unauthorized")
 53          self.status_code = 401
 54  
 55  
 56  class _FakeOpenAI:
 57      def __init__(self, **kwargs):
 58          self.kwargs = kwargs
 59  
 60      def close(self):
 61          return None
 62  
 63  
 64  class _Codex401ThenSuccessAgent(run_agent.AIAgent):
 65      refresh_attempts = 0
 66      last_init = {}
 67  
 68      def __init__(self, *args, **kwargs):
 69          kwargs.setdefault("skip_context_files", True)
 70          kwargs.setdefault("skip_memory", True)
 71          kwargs.setdefault("max_iterations", 4)
 72          type(self).last_init = dict(kwargs)
 73          super().__init__(*args, **kwargs)
 74          self._cleanup_task_resources = lambda task_id: None
 75          self._persist_session = lambda messages, history=None: None
 76          self._save_trajectory = lambda messages, user_message, completed: None
 77          self._save_session_log = lambda messages: None
 78  
 79      def _try_refresh_codex_client_credentials(self, *, force: bool = True) -> bool:
 80          type(self).refresh_attempts += 1
 81          return True
 82  
 83      def run_conversation(self, user_message: str, conversation_history=None, task_id=None):
 84          calls = {"api": 0}
 85  
 86          def _fake_api_call(api_kwargs):
 87              calls["api"] += 1
 88              if calls["api"] == 1:
 89                  raise _UnauthorizedError()
 90              return _codex_message_response("Recovered via refresh")
 91  
 92          self._interruptible_api_call = _fake_api_call
 93          return super().run_conversation(user_message, conversation_history=conversation_history, task_id=task_id)
 94  
 95  
 96  def test_cron_run_job_codex_path_handles_internal_401_refresh(monkeypatch):
 97      _patch_agent_bootstrap(monkeypatch)
 98      monkeypatch.setattr(run_agent, "OpenAI", _FakeOpenAI)
 99      monkeypatch.setattr(run_agent, "AIAgent", _Codex401ThenSuccessAgent)
100      monkeypatch.setattr(
101          "hermes_cli.runtime_provider.resolve_runtime_provider",
102          lambda requested=None: {
103              "provider": "openai-codex",
104              "api_mode": "codex_responses",
105              "base_url": "https://chatgpt.com/backend-api/codex",
106              "api_key": "codex-token",
107          },
108      )
109      monkeypatch.setattr("hermes_cli.runtime_provider.format_runtime_provider_error", lambda exc: str(exc))
110  
111      _Codex401ThenSuccessAgent.refresh_attempts = 0
112      _Codex401ThenSuccessAgent.last_init = {}
113  
114      success, output, final_response, error = cron_scheduler.run_job(
115          {"id": "job-1", "name": "Codex Refresh Test", "prompt": "ping", "model": "gpt-5.3-codex"}
116      )
117  
118      assert success is True
119      assert error is None
120      assert final_response == "Recovered via refresh"
121      assert "Recovered via refresh" in output
122      assert _Codex401ThenSuccessAgent.refresh_attempts == 1
123      assert _Codex401ThenSuccessAgent.last_init["provider"] == "openai-codex"
124      assert _Codex401ThenSuccessAgent.last_init["api_mode"] == "codex_responses"
125  
126  
127  def test_gateway_run_agent_codex_path_handles_internal_401_refresh(monkeypatch):
128      _patch_agent_bootstrap(monkeypatch)
129      monkeypatch.setattr(run_agent, "OpenAI", _FakeOpenAI)
130      monkeypatch.setattr(run_agent, "AIAgent", _Codex401ThenSuccessAgent)
131      monkeypatch.setattr(
132          gateway_run,
133          "_resolve_runtime_agent_kwargs",
134          lambda: {
135              "provider": "openai-codex",
136              "api_mode": "codex_responses",
137              "base_url": "https://chatgpt.com/backend-api/codex",
138              "api_key": "codex-token",
139          },
140      )
141      monkeypatch.setenv("HERMES_TOOL_PROGRESS", "false")
142      monkeypatch.setenv("HERMES_MODEL", "gpt-5.3-codex")
143  
144      _Codex401ThenSuccessAgent.refresh_attempts = 0
145      _Codex401ThenSuccessAgent.last_init = {}
146  
147      runner = gateway_run.GatewayRunner.__new__(gateway_run.GatewayRunner)
148      runner.adapters = {}
149      runner._ephemeral_system_prompt = ""
150      runner._prefill_messages = []
151      runner._reasoning_config = None
152      runner._provider_routing = {}
153      runner._fallback_model = None
154      runner._running_agents = {}
155      from unittest.mock import MagicMock, AsyncMock
156      runner.hooks = MagicMock()
157      runner.hooks.emit = AsyncMock()
158      runner.hooks.loaded_hooks = []
159      runner._session_db = None
160      # Ensure model resolution returns the codex model even if xdist
161      # leaked env vars cleared HERMES_MODEL.
162      monkeypatch.setattr(
163          gateway_run.GatewayRunner,
164          "_resolve_turn_agent_config",
165          lambda self, msg, model, runtime: {
166              "model": model or "gpt-5.3-codex",
167              "runtime": runtime,
168          },
169      )
170  
171      source = SessionSource(
172          platform=Platform.LOCAL,
173          chat_id="cli",
174          chat_name="CLI",
175          chat_type="dm",
176          user_id="user-1",
177      )
178  
179      result = asyncio.run(
180          runner._run_agent(
181              message="ping",
182              context_prompt="",
183              history=[],
184              source=source,
185              session_id="session-1",
186              session_key="agent:main:local:dm",
187          )
188      )
189  
190      assert result["final_response"] == "Recovered via refresh"
191      assert _Codex401ThenSuccessAgent.refresh_attempts == 1
192      assert _Codex401ThenSuccessAgent.last_init["provider"] == "openai-codex"
193      assert _Codex401ThenSuccessAgent.last_init["api_mode"] == "codex_responses"