Real model fallback retry (opt-in, at most once)

Worker previously only recorded a model_fallback event on a fallbackable failure and failed the task; ModelRouter.next_after was never used to actually run the task on another model. With --fallback-models (default empty = off, old behaviour unchanged), a fallbackable failure (404/429/5xx/timeout; 401/400 and cancellations never fall back) now re-runs the same task, same worktree/branch, on the first fallback model via host.py's top-level --model flag (no host.py signature change). The second attempt's result is final: actual_model is the fallback model, fallback_count=1, and a double failure keeps both reasons in error_text. LocalExecutor/DockerExecutor pass task model through to host.py; CloudExecutor includes model only when set.
2026-10-26 05:23:52 +08:00
1 parent f1d42d04f3
commit 7ce603b610
@@ -29,11 +29,16 @@
   - cost.py CostStore：每轮实际执行记一条 source="estimated"
     的成本记录（host 协议不暴露真实 token 数，估算绝不冒充
     actual；total_tokens 由 CostStore.record 按 input+output 算）；
-   - models.py：可 fallback 的失败（如 host 状态 timeout）且
-     router 给出下一个模型时，用 record_fallback() 在任务记录上
-     记 model_fallback 事件，并把 fallback_count 写进成本记录。
-     注意：worker 只记录 fallback 事件，不自动用下一个模型重试——
-     自动重试等于未经批准烧 relay 额度，重试与否由调用方决定。
+   - models.py：可 fallback 的失败（404/429/5xx/timeout，401/400
+     不算）。默认（未配 --fallback-models）时 worker 只记录
+     fallback 事件（record_fallback() 记 model_fallback 事件，
+     fallback_count 写进成本记录），不自动换模型重试——自动重试
+     等于未经批准烧 relay 额度。显式配了 --fallback-models 时，
+     第一次失败且可 fallback 就真的换列表里的第一个模型再跑一次
+     （最多 1 次，同一任务、同一 worktree/分支，dispatch 报
+     “任务已存在”时走现有的 run 续跑路径，模型经 host.py 的
+     顶层 --model 参数传过去，不改 host.py 签名）；第二次的结果
+     就是最终结果，actual_model 写成最后跑的模型，fallback_count=1。
   审计：认领/完成/失败/取消/缺 key 各记一条 AuthStore 审计
   （user_id="system"，与 server.sweep_stale_tasks 的做法一致），
   审计失败不影响任务状态（best-effort）。
@@ -61,13 +66,21 @@ interrupted，不用执行结果覆盖它。
  的调用，超时只能靠执行器自身的 timeout，这是诚实边界，
  不要假装同步路径也能中途杀掉。
 - max_retries（默认 1 次重试，CLI --max-retries 可调，0 表示
-  不重试）是对模型任务失败的有限重试：取消不重试；已记
-  fallback 事件的失败不做同模型重试（fallback 本身就是模型
-  级的重试机制，再同模型重试会重复烧额度）；其它失败（含
-  worker 超时）重试前按 retry_backoff_seconds（默认 5 秒，
+  不重试）是对模型任务失败的有限重试：取消不重试；可 fallback
+  的失败优先走真实 fallback（见下），不再做同模型重试（fallback
+  本身就是模型级的重试，再同模型重试会重复烧额度）；其它失败
+  （含 worker 超时）重试前按 retry_backoff_seconds（默认 5 秒，
  测试可注入 0）等待，重试次数记在审计（worker_retry）与
  最终 error_text（[retries=N] 前缀）里，不另起新表；重试
  耗尽后任务才最终 failed，成本记录与 usage_turns 只记一次
+- 真实 fallback（CLI --fallback-models，默认空 = 关闭）：一次
+  尝试失败且可 fallback（404/429/5xx/timeout，401/400 不算，
+  取消不算）时，若配了 fallback 列表，就记 model_fallback 事件
+  与 worker_fallback 审计，然后换列表里的第一个模型再跑一次
+  （最多 1 次，与 max_retries 无关，max_retries=0 时 fallback
+  照样跑）；第二次的结果就是最终结果，actual_model 写成最后
+  跑的模型，fallback_count=1，两次都失败时 error_text 同时写清
+  两次原因；未配列表时保持老行为（只记事件、直接 failed）。
  （按最终结果记，不按尝试次数重复记）。
 - 容器清理兜底：Worker.cleanup_stale_containers() 列出
  codex- 前缀的残留容器并逐个 kill+rm，失败只记日志不炸
@@ -219,7 +232,7 @@ class Worker:
                 task_timeout=1800, max_retries=1,
                 retry_backoff_seconds=5.0, repos_dir="./repos",
                 repos_runner=None, heartbeat_file=None,
-                 worker_id=None, capacity=1):
+                 worker_id=None, capacity=1, fallback_models=None):
        self.db_path = str(db_path)
        self.host_py_path = host_py_path
        self.config_path = config_path
@@ -232,6 +245,19 @@ class Worker:
        if router is not None and not isinstance(router, ModelRouter):
            raise ValueError("router 必须是 cloud.models.ModelRouter")
        self.router = router
+        # 真实 fallback 换模型列表：默认空 = 关闭，只记事件不换模型
+        # （老行为，生产 worker 未配该参数时绝不突然多花模型钱）。
+        # 归一化：去空白、去重保序、去掉与主模型相同的项（换成同一
+        # 个模型不叫 fallback，只会白烧一次额度）；fallback 总次数
+        # 最多 1 次，所以运行期只用列表的第一个元素，后面的只是
+        # 备用声明，不会被依次试下去。
+        self.fallback_models = []
+        for _m in (fallback_models or []):
+            if not isinstance(_m, str):
+                continue
+            _m = _m.strip()
+            if _m and _m != self.model and _m not in self.fallback_models:
+                self.fallback_models.append(_m)
        self.max_turns = max_turns
        self.max_cost_usd = max_cost_usd
        self.usage_turns = 0
@@ -860,6 +886,29 @@ class Worker:
            result = dict(result, status="timeout")
        return result, error

+    def _attempt_error(self, result, error, token):
+        """一次尝试失败的可读原因（供 fallback 两次原因拼接用）。
+
+        超时用统一的超时文本；runner 异常用异常文本；同步路径的
+        stdout 是 host.py 的 JSON 时取它的 text 字段（比整段 JSON
+        可读）；最后退到 host_status token 或通用文本。绝不返回
+        任何密钥（stdout 里本就不应含密钥，text 字段同样不含）。
+        """
+        if result is not None and result.get("status") == "timeout":
+            return "task timeout after %s seconds" % self.task_timeout
+        if error:
+            return error
+        stdout = (result or {}).get("stdout", "") or ""
+        try:
+            parsed = json.loads(stdout)
+        except (TypeError, ValueError):
+            parsed = None
+        if isinstance(parsed, dict) and parsed.get("text"):
+            return str(parsed["text"])
+        if token is not None:
+            return "host_status=%s" % token
+        return stdout.strip() or "host execution failed"
+
    def run_one(self, task=None):
        """认领并执行一个任务，返回结果 dict（见模块 docstring）。

@@ -928,9 +977,24 @@ class Worker:
        result = None
        error = None
        retry_count = 0
-        # 有限重试：取消不重试；已记 fallback 的失败不做同模型重试
-        # （fallback 本身就是模型级重试机制）；其它失败（含 worker
-        # 超时）按 max_retries 重试，退避等待后重跑一次 _run_attempt。
+        # 最后真正跑过的模型：主模型起手，fallback 后换成 fallback 模型。
+        actual_model = self.model
+        fallback_model = None
+        first_error = None
+        # 模型经任务的 model 字段传给执行器（执行器再经 host.py 的
+        # 顶层 --model 参数传给 dispatch/run）；不设时 host.py 用
+        # 配置文件里的 model。这里显式设成 worker 的主模型，让
+        # --model 覆盖配置真正生效，fallback 时再换成新模型。
+        task = dict(task)
+        task["model"] = self.model
+        # 有限重试与真实 fallback 的优先级：取消不重试；一次尝试失败
+        # 且可 fallback、且显式配了 fallback_models 时，优先真换模型
+        # 再跑一次（最多 1 次，同一任务、同一 worktree/分支，dispatch
+        # 报“任务已存在”时 _run_attempt 内部走现有的 run 续跑路径），
+        # 第二次的结果就是最终结果，不再做同模型重试；未配
+        # fallback_models 时保持老行为（router 给出下一模型时只记事件
+        # 不重试）；其它失败（含 worker 超时）按 max_retries 做同模型
+        # 重试，退避等待后重跑一次 _run_attempt。
        for attempt in range(self.max_retries + 1):
            result, error = self._run_attempt(task)
            if self._was_cancelled(task_id) or \
@@ -939,6 +1003,32 @@ class Worker:
            if result.get("status") == "completed":
                break
            host_status_try = result.get("host_status")
+            if host_status_try is None and \
+                    result.get("status") == "timeout":
+                # 进程/容器超时路径的 host_status 为空，超时本身就是
+                # 可 fallback 的失败 token（models.FALLBACKABLE 含它）。
+                host_status_try = "timeout"
+            if fallback_model is None and self.fallback_models \
+                    and host_status_try is not None \
+                    and is_fallbackable(host_status_try):
+                fb = self.fallback_models[0]
+                record = dict(task)
+                record["events"] = events
+                # record_fallback 本身会 append 到 events，
+                # 这里不要再 append 一次，否则事件会重复。
+                record_fallback(
+                    record, self.model, fb, str(host_status_try))
+                fallback_count = 1
+                self._audit(task["project_id"], "worker_fallback",
+                            "%s -> %s" % (self.model, fb))
+                first_error = self._attempt_error(
+                    result, error, host_status_try)
+                fallback_model = fb
+                actual_model = fb
+                task = dict(task)
+                task["model"] = fb
+                result, error = self._run_attempt(task)
+                break
            if self.router is not None and host_status_try is not None \
                    and is_fallbackable(host_status_try) and \
                    self.router.next_after(
@@ -968,7 +1058,7 @@ class Worker:
            self._set_status(task_id, "interrupted")
            self._write_execution_result(
                task_id, result_text=(result or {}).get("stdout", ""),
-                error_text="cancelled", actual_model=self.model,
+                error_text="cancelled", actual_model=actual_model,
                started_at=started_at, finished_at=time.time())
            log_event(self.logger, "worker_cancelled", task_id=task_id,
                      project_id=task["project_id"], status="interrupted")
@@ -984,36 +1074,48 @@ class Worker:
        host_status = result.get("host_status")
        final = result["status"]
        if final != "completed":
-            if error is None:
-                error = stdout.strip() or "host execution failed"
-            if retry_count:
-                error = "[retries=%d] %s" % (retry_count, error)
-            # fallback 事件：只记、不自动重试（见模块 docstring）。
-            if self.router is not None and host_status is not None \
-                    and is_fallbackable(host_status):
-                nxt = self.router.next_after(self.model, host_status)
-                if nxt is not None:
-                    record = dict(task)
-                    record["events"] = events
-                    # record_fallback 本身会 append 到 events，
-                    # 这里不要再 append 一次，否则事件会重复。
-                    record_fallback(
-                        record, self.model, nxt, str(host_status))
-                    fallback_count = 1
-                    self._audit(task["project_id"], "worker_fallback",
-                                "%s -> %s" % (self.model, nxt))
+            if fallback_model is not None:
+                # 真 fallback 后第二次也失败：错误文本必须同时写清
+                # 两次原因，不能只留第二次的（超时转换已在前面把
+                # error 置成超时文本，那时 second_error 就是它）。
+                second_error = error or self._attempt_error(
+                    result, None, host_status)
+                error = ("primary model %s failed: %s; "
+                         "fallback model %s failed: %s"
+                         % (self.model, first_error,
+                            fallback_model, second_error))
+            else:
+                if error is None:
+                    error = stdout.strip() or "host execution failed"
+                if retry_count:
+                    error = "[retries=%d] %s" % (retry_count, error)
+                # 未配 fallback_models 时的老行为：只记事件、不重试。
+                if self.router is not None and host_status is not None \
+                        and is_fallbackable(host_status):
+                    nxt = self.router.next_after(self.model, host_status)
+                    if nxt is not None:
+                        record = dict(task)
+                        record["events"] = events
+                        # record_fallback 本身会 append 到 events，
+                        # 这里不要再 append 一次，否则事件会重复。
+                        record_fallback(
+                            record, self.model, nxt, str(host_status))
+                        fallback_count = 1
+                        self._audit(task["project_id"], "worker_fallback",
+                                    "%s -> %s" % (self.model, nxt))
        self._set_status(task_id, final)
        finished_at = time.time()
        self._write_execution_result(
            task_id, result_text=stdout, error_text=error,
-            actual_model=self.model, started_at=started_at,
+            actual_model=actual_model, started_at=started_at,
            finished_at=finished_at)
        self.usage_turns += 1
        # 成本：host 协议不暴露真实 token 数，用 cost.py 的估算结构
-        # 记 estimated，绝不标 actual；fallback_count 复用同一记录。
+        # 记 estimated，绝不标 actual；fallback_count 复用同一记录；
+        # 模型记最后真正跑的那个（fallback 成功时就是 fallback 模型）。
        total = estimate_tokens(task.get("prompt") or "", stdout)
        cost_id = self.costs.record(
-            task_id, task["project_id"], self.model,
+            task_id, task["project_id"], actual_model,
            input_tokens=total, output_tokens=0, usd_micros=None,
            source="estimated", fallback_count=fallback_count)
        self._audit(task["project_id"],
@@ -1029,6 +1131,7 @@ class Worker:
                if hasattr(self.executor, "_host_name") else task_id,
                "status": final, "executor": self._executor_name(),
                "stdout": stdout, "error": error, "events": events,
+                "actual_model": actual_model,
                "cost_record_id": cost_id}

    def run_all(self, max_tasks=None):
@@ -1157,6 +1260,12 @@ def main(argv=None):
    parser.add_argument("--codex-config", default=None,
                        help="含 env_key 行的 codex 配置文件（只读名字）")
    parser.add_argument("--model", default=None)
+    parser.add_argument("--fallback-models", default=None,
+                        help="真实 fallback 换模型列表，逗号分隔"
+                             "（默认空 = 关闭，只记事件不换模型；"
+                             "配了时主模型可 fallback 失败后最多换"
+                             "列表里的第一个模型再跑一次，401/400"
+                             "绝不 fallback，取消绝不 fallback）")
    parser.add_argument("--max-turns", type=int, default=None)
    parser.add_argument("--max-tasks", type=int, default=None)
    parser.add_argument("--executor", default="local",
@@ -1205,6 +1314,10 @@ def main(argv=None):
    worker = Worker(args.db, host_py, config_path=args.config,
                    repo=args.repo, env_key=args.env_key,
                    codex_config_path=args.codex_config, model=args.model,
+                    fallback_models=(
+                        [m.strip() for m in
+                         args.fallback_models.split(",")]
+                        if args.fallback_models else []),
                    max_turns=args.max_turns,
                    executor=_build_cli_executor(args, host_py),
                    task_timeout=args.task_timeout,