diff --git a/cloud/tests/test_fallback_retry.py b/cloud/tests/test_fallback_retry.py index 0bdde5e..b66e746 100644 --- a/cloud/tests/test_fallback_retry.py +++ b/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() diff --git a/cloud/worker.py b/cloud/worker.py index 7a242e6..919863e 100644 --- a/cloud/worker.py +++ b/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 为空,超时本身就是