Merge pull request #13 from openqj/fallback-thread-start-crash

Fallback also triggers on thread/start crash with explicit HTTP code

cloud/tests/test_fallback_retry.py

@@ -261,6 +261,69 @@ class FallbackRetryTestCase(unittest.TestCase):
         self.assertEqual(len(self.calls), 2)
         self.assertEqual(result["events"][0]["reason"], "timeout")
 
+    def test_thread_start_crash_with_404_in_stderr_fallback_succeeds(self):
+        # 坏模型最常见的真实失败：thread/start 抛错、host.py 非 0
+        # 退出、不打印 JSON，host_status 为 None；stderr 里有显式
+        # 404 时必须 fallback，第二次换模型成功（主机实测回归）。
+        def fake(cmd):
+            if _model_of(cmd) == "gpt-6-sol":
+                return _host_json("A", "completed", "done by fallback")
+            return SimpleNamespace(
+                returncode=1, stdout="",
+                stderr="RuntimeError: thread/start 失败: {'error': "
+                       "{'code': 404, 'message': 'model not found'}}")
+
+        _insert_task(self.db_path, "A")
+        worker = self._worker(
+            runner=self._recording(fake),
+            model="gpt-6.1-sol",
+            fallback_models=["gpt-6-sol"],
+            max_retries=0)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["actual_model"], "gpt-6-sol")
+        self.assertEqual(len(self.calls), 2)
+        self.assertEqual(result["events"][0]["reason"], "404")
+        row, cost, audits = _execution(self.db_path, "A")
+        self.assertEqual(cost[1], 1)
+        self.assertIn("worker_fallback", audits)
+
+    def test_thread_start_crash_with_401_in_stderr_never_fallbacks(self):
+        # stderr 里是 401 时绝不 fallback，只调一次。
+        def fake(cmd):
+            return SimpleNamespace(
+                returncode=1, stdout="",
+                stderr="RuntimeError: thread/start 失败: {'error': "
+                       "{'code': 401, 'message': 'unauthorized'}}")
+
+        _insert_task(self.db_path, "A")
+        worker = self._worker(
+            runner=self._recording(fake),
+            model="gpt-6.1-sol",
+            fallback_models=["gpt-6-sol"],
+            max_retries=0)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual(len(self.calls), 1)
+        self.assertEqual(result["events"], [])
+
+    def test_thread_start_crash_without_code_never_fallbacks(self):
+        # 没有显式状态码的通用失败不猜测：不 fallback，只调一次。
+        def fake(cmd):
+            return SimpleNamespace(
+                returncode=1, stdout="", stderr="host execution failed")
+
+        _insert_task(self.db_path, "A")
+        worker = self._worker(
+            runner=self._recording(fake),
+            model="gpt-6.1-sol",
+            fallback_models=["gpt-6-sol"],
+            max_retries=0)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual(len(self.calls), 1)
+        self.assertEqual(result["events"], [])
+
 
 if __name__ == "__main__":
     unittest.main()

cloud/worker.py

@@ -909,6 +909,44 @@ class Worker:
             return "host_status=%s" % token
         return stdout.strip() or "host execution failed"
 
+    def _fallback_token(self, result, error):
+        """一次失败尝试的可 fallback 信号（host_status 为空时的兜底）。
+
+        坏模型最常见的真实失败不在 turn 里，而在 thread/start：
+        host.py 的 create_thread 抛 RuntimeError、进程非 0 退出、
+        不打印 JSON，于是执行器结果的 host_status 为 None，旧逻辑
+        因此永远不 fallback（主机实测：坏模型任务直接 failed、
+        无 worker_fallback 审计）。此时从 stdout/stderr/runner 异常
+        文本里找显式 HTTP 状态码（404/429/5xx）或“model not found”
+        类文本，前者按码判断，后者按 404 计；找不到显式信号时
+        返回 None，绝不凭“host execution failed”这类通用文本猜测，
+        否则任何执行器故障都会被误当成模型故障换模型烧额度。
+        """
+        host_status = (result or {}).get("host_status")
+        if host_status is not None:
+            return host_status
+        combined = " ".join([
+            (result or {}).get("stdout", "") or "",
+            (result or {}).get("stderr", "") or "",
+            error or "",
+        ])
+        if not combined:
+            return None
+        lowered = combined.lower()
+        if "model not found" in lowered or "unknown model" in lowered \
+                or "no such model" in lowered:
+            return "404"
+        match = re.search(
+            r"(?:http/|status[\"':\s]{0,10}|code[\"':\s]{0,10}|error[\"':\s]{0,10})"
+            r"(404|429|500|502|503|504)\b",
+            combined, re.IGNORECASE)
+        if match:
+            return match.group(1)
+        match = re.search(r"\b(404|429|500|502|503|504)\b", combined)
+        if match:
+            return match.group(1)
+        return None
+
     def run_one(self, task=None):
         """认领并执行一个任务，返回结果 dict（见模块 docstring）。
 
@@ -1002,7 +1040,7 @@ class Worker:
                 break
             if result.get("status") == "completed":
                 break
-            host_status_try = result.get("host_status")
+            host_status_try = self._fallback_token(result, error)
             if host_status_try is None and \
                     result.get("status") == "timeout":
                 # 进程/容器超时路径的 host_status 为空，超时本身就是