refactor(chatgpt): enforce browser->vm sentinel fallback order

platforms/chatgpt/refresh_token_registration_engine.py

@@ -19,7 +19,7 @@ from core.task_runtime import TaskInterruption
 from .oauth import OAuthManager, OAuthStart
 from .http_client import OpenAIHTTPClient
 from .sentinel_browser import get_sentinel_token_via_browser
-from .sentinel_token import build_sentinel_token
+from .sentinel_token import build_sentinel_token, build_sentinel_token_vm_only
 from .utils import (
     generate_datadog_trace,
     generate_device_id,
@@ -349,6 +349,14 @@ class RefreshTokenRegistrationEngine:
                 if browser_token:
                     self._log(f"Sentinel Browser token 获取成功 ({flow})")
                     return browser_token
+                self._log(f"Sentinel Browser token 获取失败，尝试 VM ({flow})", "warning")
+                vm_token = build_sentinel_token_vm_only(self.session, did, flow=flow)
+                if vm_token:
+                    self._log(f"Sentinel VM token 获取成功 ({flow})")
+                    return vm_token
+                self._log(f"Sentinel VM token 获取失败 ({flow})", "warning")
+                return None
+
             sen_token = build_sentinel_token(self.session, did, flow=flow)
             if sen_token:
                 self._log(f"Sentinel token 获取成功 ({flow})")

platforms/chatgpt/sentinel_token.py

@@ -421,3 +421,22 @@ def build_sentinel_token(session, device_id, flow="authorize_continue", user_age
         "id": device_id,
         "flow": flow,
     }, separators=(",", ":"))
+
+
+def build_sentinel_token_vm_only(
+    session,
+    device_id,
+    flow="authorize_continue",
+    user_agent=None,
+    sec_ch_ua=None,
+    impersonate=None,
+):
+    """仅使用 VM 链路生成 sentinel token（不回退 PoW）。"""
+    return _build_sentinel_token_via_vm(
+        session,
+        device_id,
+        flow=flow,
+        user_agent=user_agent,
+        sec_ch_ua=sec_ch_ua,
+        impersonate=impersonate,
+    )

tests/test_chatgpt_register.py

@@ -328,23 +328,23 @@ class RegistrationEngineFlowTests(unittest.TestCase):
         mock_pow_token.assert_not_called()
 
     @mock.patch(
-        "platforms.chatgpt.refresh_token_registration_engine.build_sentinel_token",
-        return_value='{"source":"pow"}',
+        "platforms.chatgpt.refresh_token_registration_engine.build_sentinel_token_vm_only",
+        return_value='{"source":"vm"}',
     )
     @mock.patch(
         "platforms.chatgpt.refresh_token_registration_engine.get_sentinel_token_via_browser",
         return_value=None,
     )
-    def test_check_sentinel_falls_back_to_pow_when_browser_token_missing(
-        self, mock_browser_token, mock_pow_token
+    def test_check_sentinel_falls_back_to_vm_when_browser_token_missing(
+        self, mock_browser_token, mock_vm_token
     ):
         engine = self._make_engine()
         engine.session = mock.Mock()
 
         token = engine._check_sentinel("device-fixed", flow="oauth_create_account")
-        self.assertEqual(token, '{"source":"pow"}')
+        self.assertEqual(token, '{"source":"vm"}')
         mock_browser_token.assert_called_once()
-        mock_pow_token.assert_called_once_with(
+        mock_vm_token.assert_called_once_with(
             engine.session, "device-fixed", flow="oauth_create_account"
         )