fix: detect add-phone oauth blocks earlier

src/autoteam/codex_auth.py

@@ -35,6 +35,7 @@
 CODEX_TOKEN_URL = "https://auth.openai.com/oauth/token"
 CODEX_CALLBACK_PORT = 1455
 CODEX_REDIRECT_URI = f"http://localhost:{CODEX_CALLBACK_PORT}/auth/callback"
+_EARLY_OAUTH_BLOCK_FAILURE_TYPES = {"add_phone", "human_verification", "site_unavailable"}
 
 
 def _generate_pkce():
@@ -93,6 +94,48 @@ def _classify_oauth_failure(url, body_excerpt=""):
     return "auth_code_missing", f"未获取到 auth code（停留在 {url or 'unknown'}）", True
 
 
+def _build_oauth_failure_result(url, body_excerpt=""):
+    error_type, error_detail, retryable = _classify_oauth_failure(url, body_excerpt)
+    return {
+        "ok": False,
+        "bundle": None,
+        "error_type": error_type,
+        "error_detail": error_detail,
+        "retryable": retryable,
+        "current_url": url,
+        "body_excerpt": body_excerpt,
+    }
+
+
+def _detect_early_oauth_block(page):
+    body_excerpt = _page_excerpt(page)
+    failure = _build_oauth_failure_result(getattr(page, "url", ""), body_excerpt)
+    if failure["error_type"] in _EARLY_OAUTH_BLOCK_FAILURE_TYPES:
+        return failure
+    return None
+
+
+def _wait_for_oauth_page_progress(page, previous_url="", timeout=5):
+    previous_url = (previous_url or "").lower()
+    deadline = time.time() + timeout
+
+    while time.time() < deadline:
+        current_url = (getattr(page, "url", "") or "").lower()
+        if f"localhost:{CODEX_CALLBACK_PORT}/auth/callback" in current_url:
+            return None
+
+        failure = _detect_early_oauth_block(page)
+        if failure:
+            return failure
+
+        if previous_url and current_url and current_url != previous_url:
+            return None
+
+        time.sleep(0.5)
+
+    return _detect_early_oauth_block(page)
+
+
 def _build_auth_url(code_challenge, state):
     params = {
         "client_id": CODEX_CLIENT_ID,
@@ -1071,6 +1114,18 @@ def on_response(response):
             if auth_code:
                 break
 
+            blocking_failure = _detect_early_oauth_block(page)
+            if blocking_failure:
+                logger.warning(
+                    "[Codex] 提前识别到 OAuth 阻塞页 (step %d): %s | URL=%s",
+                    step + 1,
+                    blocking_failure["error_detail"],
+                    blocking_failure["current_url"],
+                )
+                _screenshot(page, f"codex_04_blocked_{step + 1}.png")
+                failure_result = blocking_failure
+                break
+
             _screenshot(page, f"codex_04_step{step + 1}_before.png")
 
             try:
@@ -1103,11 +1158,24 @@ def on_response(response):
                         try:
                             cont_btn = page.locator('button:has-text("继续"), button:has-text("Continue")').first
                             if cont_btn.is_visible(timeout=3000):
+                                previous_url = page.url
                                 cont_btn.click()
-                                time.sleep(3)
+                                failure = _wait_for_oauth_page_progress(page, previous_url=previous_url, timeout=3)
+                                if failure:
+                                    logger.warning(
+                                        "[Codex] 提前识别到 OAuth 阻塞页 (step %d): %s | URL=%s",
+                                        step + 1,
+                                        failure["error_detail"],
+                                        failure["current_url"],
+                                    )
+                                    _screenshot(page, f"codex_04_workspace_blocked_{step + 1}.png")
+                                    failure_result = failure
+                                    break
                                 logger.info("[Codex] 已点击继续 (step %d)", step + 1)
                         except Exception:
                             pass
+                        if failure_result:
+                            break
                         continue
                     else:
                         logger.warning("[Codex] 无法选择 workspace '%s' (step %d)", workspace_name, step + 1)
@@ -1181,9 +1249,19 @@ def on_response(response):
                 ).first
                 if consent_btn.is_visible(timeout=5000):
                     logger.info("[Codex] 点击同意/继续按钮 (step %d)...", step + 1)
+                    previous_url = page.url
                     consent_btn.click()
-                    time.sleep(5)
+                    failure = _wait_for_oauth_page_progress(page, previous_url=previous_url, timeout=5)
                     _screenshot(page, f"codex_04_consent_{step + 1}.png")
+                    if failure:
+                        logger.warning(
+                            "[Codex] 提前识别到 OAuth 阻塞页 (step %d): %s | URL=%s",
+                            step + 1,
+                            failure["error_detail"],
+                            failure["current_url"],
+                        )
+                        failure_result = failure
+                        break
                 else:
                     time.sleep(1)
                     continue
@@ -1213,16 +1291,7 @@ def on_response(response):
             _screenshot(page, "codex_05_no_callback.png")
             body_excerpt = _page_excerpt(page)
             logger.warning("[Codex] 未获取到 auth code，当前 URL: %s", page.url)
-            error_type, error_detail, retryable = _classify_oauth_failure(page.url, body_excerpt)
-            failure_result = {
-                "ok": False,
-                "bundle": None,
-                "error_type": error_type,
-                "error_detail": error_detail,
-                "retryable": retryable,
-                "current_url": page.url,
-                "body_excerpt": body_excerpt,
-            }
+            failure_result = _build_oauth_failure_result(page.url, body_excerpt)
 
         browser.close()
 
@@ -1392,6 +1461,10 @@ def _detect_step(self):
             if self.auth_code:
                 return "completed", None
 
+        blocking_failure = _detect_early_oauth_block(self.page)
+        if blocking_failure:
+            return blocking_failure["error_type"], blocking_failure["error_detail"]
+
         if self._visible_locator(self.CODE_SELECTORS, timeout_ms=800):
             return "code_required", None
         if self._visible_locator(self.PASSWORD_SELECTORS, timeout_ms=800):
@@ -1561,6 +1634,9 @@ def _advance(self, attempts=12):
                     "detail": "主号 Codex 当前停留在密码页，且未找到一次性验证码入口",
                 }
 
+            if step in _EARLY_OAUTH_BLOCK_FAILURE_TYPES:
+                return {"step": step, "detail": detail}
+
             if step == "email_required":
                 if self._auto_fill_email():
                     continue

tests/unit/test_codex_auth_session.py

@@ -84,10 +84,11 @@ def test_refresh_main_auth_file_saves_bundle_from_session_login(monkeypatch):
 
 
 class _FakeElement:
-    def __init__(self, text, *, visible=True):
+    def __init__(self, text, *, visible=True, on_click=None):
         self._text = text
         self.visible = visible
         self.clicked = False
+        self._on_click = on_click
 
     def is_visible(self, timeout=0):
         return self.visible
@@ -97,6 +98,8 @@ def inner_text(self, timeout=0):
 
     def click(self, timeout=0, force=False):
         self.clicked = True
+        if self._on_click:
+            self._on_click()
 
 
 class _FakeCollection:
@@ -185,6 +188,31 @@ def advance(self):
         self._body = "Codex wants access to your API organization Select a project Continue"
 
 
+class _FakeConsentToAddPhonePage(_FakePage):
+    def __init__(self):
+        self._continue_button = _FakeElement("Continue", on_click=self._advance_to_add_phone)
+        super().__init__(
+            url="https://auth.openai.com/sign-in-with-chatgpt/codex/consent",
+            body="Codex wants access to your API organization Select a project Continue",
+            elements=[],
+        )
+
+    def _advance_to_add_phone(self):
+        self.url = "https://auth.openai.com/add-phone"
+        self._body = "Add a phone number to continue"
+
+    def locator(self, selector):
+        if selector == "body":
+            return _FakeCollection(text=self._body)
+        if selector in {
+            'button:has-text("Continue"), button:has-text("继续"), button:has-text("Allow")',
+        }:
+            return _FakeCollection(items=[self._continue_button])
+        if selector in self._GENERIC_SELECTORS:
+            return _FakeCollection(items=self._elements)
+        return _FakeCollection(items=[])
+
+
 def test_workspace_selection_detection_ignores_otp_pages():
     page = _FakePage(
         url="https://auth.openai.com/email-verification",
@@ -393,3 +421,41 @@ def test_team_workspace_selection_requires_exact_workspace_name():
 
     assert codex_auth._workspace_label_candidates(page) == [("Personal account", items[1])]
     assert codex_auth._select_team_workspace(page, "Idapro") is False
+
+
+def test_detect_early_oauth_block_detects_add_phone_page():
+    page = _FakePage(url="https://auth.openai.com/add-phone", body="Add a phone number to continue")
+
+    failure = codex_auth._detect_early_oauth_block(page)
+
+    assert failure is not None
+    assert failure["error_type"] == "add_phone"
+    assert failure["error_detail"] == "需要手机号验证"
+    assert failure["retryable"] is False
+
+
+def test_wait_for_oauth_page_progress_detects_add_phone_after_consent_click():
+    page = _FakeConsentToAddPhonePage()
+    consent_btn = page.locator('button:has-text("Continue"), button:has-text("继续"), button:has-text("Allow")').first
+
+    previous_url = page.url
+    consent_btn.click()
+    failure = codex_auth._wait_for_oauth_page_progress(page, previous_url=previous_url, timeout=1)
+
+    assert failure is not None
+    assert failure["error_type"] == "add_phone"
+    assert failure["current_url"] == "https://auth.openai.com/add-phone"
+
+
+def test_session_codex_flow_advance_returns_add_phone_immediately():
+    flow = codex_auth.SessionCodexAuthFlow(
+        email="user@example.com",
+        session_token="session-token",
+        account_id="acc-1",
+        workspace_name="Idapro",
+    )
+    flow.page = _FakePage(url="https://auth.openai.com/add-phone", body="Add a phone number to continue")
+
+    result = flow._advance(attempts=1)
+
+    assert result == {"step": "add_phone", "detail": "需要手机号验证"}