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

Live test found the gap: a nonexistent model fails in
thread/start, host.py exits non-zero without JSON, so the
executor result has host_status=None and the worker never
fell back (task failed, no worker_fallback audit). When
host_status is absent, worker now extracts an explicit signal
from stdout/stderr/runner error: 404/429/5xx codes, or
'model not found' style text (counted as 404). Generic
failures with no explicit code still do not fall back, and
401 in stderr still never falls back. Tests: fallback 9 pass,
cloud 333 pass, host 27 pass.

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 为空，超时本身就是