Merge pull request #12 from openqj/model-fallback-retry

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

cloud/README.md

@@ -530,11 +530,27 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   执行器自身的 timeout，这是诚实边界，不要假装同步路径也能中途杀掉。
 - 失败重试与退避：worker 的 `--max-retries`（默认 1 次重试，0 表示
   不重试）与 `--retry-backoff-seconds`（默认 5 秒，测试可注入 0）；
-  取消不重试，已记 fallback 事件的失败不做同模型重试（fallback 本身
-  就是模型级的重试机制），其它失败（含 worker 超时）重试前按固定
-  退避等待；重试次数记在审计（`worker_retry`）与最终 `error_text`
+  取消不重试，其它失败（含 worker 超时）重试前按固定退避等待；
+  重试次数记在审计（`worker_retry`）与最终 `error_text`
   的 `[retries=N]` 前缀里，不另起新表；重试耗尽后任务才最终 failed，
   成本记录与 usage_turns 只按最终结果记一次，不按尝试次数重复记。
+- 模型 fallback（显式配置时真换模型重试一次）：worker 的
+  `--fallback-models`（逗号分隔的模型列表，默认空 = 关闭）。
+  未配时保持老行为：可 fallback 的失败只记 `model_fallback` 事件
+  与 `worker_fallback` 审计、`fallback_count=1`，然后直接 failed，
+  不换模型。显式配了时，一次尝试失败且错误可 fallback（404/429/
+  5xx/timeout；401/400 绝不 fallback，取消绝不 fallback），就先记
+  事件与审计，再用列表里的第一个模型在同一任务、同一 worktree/分支
+  上再跑一次（dispatch 报“任务已存在”时走现有的 run 续跑路径，
+  模型经 host.py 的顶层 `--model` 参数传过去，不改 host.py 签名）；
+  fallback 总次数最多 1 次，配两个模型也只试第一个，不会无限换下去
+  烧额度，且与 `--max-retries` 无关（`--max-retries 0` 时 fallback
+  照样跑）。第二次的结果就是最终结果：成功则任务 completed，
+  `actual_model` 写成 fallback 后的模型，成本记录 `fallback_count=1`；
+  第二次也失败则任务 failed，`error_text` 同时写清主模型失败原因与
+  fallback 模型失败原因。边界：真实 token 成本仍是估算（host 协议
+  不暴露真实 token 数），fallback 不改变成本来源，只改变实际跑的
+  模型与 `actual_model`/`fallback_count` 的记录。
 - 容器清理兜底：`DockerExecutor.list_containers(prefix="codex-")`（
   `docker ps -a --format {{.Names}}`，失败返回 `[]`）与
   `cleanup_stale_containers()`（列出并逐个 kill+rm，单个失败只跳过，

cloud/sandbox.py

@@ -67,16 +67,24 @@ class LocalExecutor(Executor):
     """
 
     def __init__(self, host_py_path, config_path=None, runner=None,
-                 repo=None):
+                 repo=None, model=None):
         self.host_py_path = host_py_path
         self.config_path = config_path
         self.runner = runner if runner is not None else _default_runner
         self.repo = repo
+        # 模型：host.py 的 --model 是顶层全局参数（dispatch/run 共用），
+        # 不传时 host.py 用配置文件里的 model；传了就覆盖配置。
+        # worker 的 fallback 靠它把换模型后的第二次尝试真正发到新模型上，
+        # 不改 host.py 的 dispatch/run 签名。
+        self.model = model
 
-    def _base_command(self):
+    def _base_command(self, model=None):
         cmd = [sys.executable, self.host_py_path]
         if self.config_path is not None:
             cmd += ["--config", self.config_path]
+        effective = model or self.model
+        if effective:
+            cmd += ["--model", effective]
         return cmd
 
     def _command(self, task_id):
@@ -136,17 +144,20 @@ class LocalExecutor(Executor):
     def dispatch_task(self, task) -> dict:
         """调 host.py dispatch：建 host 任务并跑一轮 turn。
 
-        命令：host.py [--config CFG] dispatch --name <host 名>
+        命令：host.py [--config CFG] [--model M] dispatch --name <host 名>
         --task <prompt> [--repo <repo>]。repo 取任务的 repo 字段，
         没有则取构造时传的 repo，都没有则不传（此时 host.py 用其
         配置文件里的 repo，配置里也没有时 host.py 自己报错，
-        worker 把该报错原样写回 failed，不编造成功）。
+        worker 把该报错原样写回 failed，不编造成功）。模型取任务的
+        model 字段，没有则取构造时传的 model，都没有则不传 --model，
+        host.py 用配置文件里的 model（fallback 时 worker 会把任务的
+        model 换成 fallback 模型，第二次 dispatch/run 就跑在新模型上）。
         """
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")
         if not task.get("prompt"):
             raise ValueError("任务缺少 prompt")
-        cmd = self._base_command()
+        cmd = self._base_command(model=task.get("model"))
         cmd += ["dispatch", "--name", self._host_name(task),
                 "--task", task["prompt"]]
         repo = task.get("repo") or self.repo
@@ -158,15 +169,16 @@ class LocalExecutor(Executor):
     def run_task(self, task) -> dict:
         """调 host.py run：在已有 host 任务上再跑一轮 turn。
 
-        命令：host.py [--config CFG] run --task <host 名>
+        命令：host.py [--config CFG] [--model M] run --task <host 名>
         --prompt <prompt>。worker 只在 dispatch 报“任务已存在”时
-        用它续跑，不用它做首次派发。
+        用它续跑，不用它做首次派发。模型与 dispatch_task 同规则：
+        任务的 model 字段优先，fallback 续跑时就是新模型。
         """
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")
         if not task.get("prompt"):
             raise ValueError("任务缺少 prompt")
-        cmd = self._base_command()
+        cmd = self._base_command(model=task.get("model"))
         cmd += ["run", "--task", self._host_name(task),
                 "--prompt", task["prompt"]]
         return self._result_from_runner(
@@ -184,7 +196,7 @@ class LocalExecutor(Executor):
             raise ValueError("任务缺少 id")
         if not task.get("prompt"):
             raise ValueError("任务缺少 prompt")
-        cmd = self._base_command()
+        cmd = self._base_command(model=task.get("model"))
         if kind == "run":
             cmd += ["run", "--task", self._host_name(task),
                     "--prompt", task["prompt"]]
@@ -251,12 +263,17 @@ class DockerExecutor(Executor):
     def __init__(self, host_py_path=None, config_path=None, runner=None,
                  image="codex-app-server:latest", cpus="2", memory="2g",
                  user="1000:1000", worktrees_root=None, env=None,
-                 env_key="XCM_API_KEY", timeout=None, extra_mounts=None):
+                 env_key="XCM_API_KEY", timeout=None, extra_mounts=None,
+                 model=None):
         if host_py_path is None:
             here = os.path.dirname(os.path.abspath(__file__))
             host_py_path = os.path.join(os.path.dirname(here), "host.py")
         self.host_py_path = host_py_path
         self.config_path = config_path
+        # 模型与 LocalExecutor 同规则：任务的 model 字段优先，没有则
+        # 取构造时传的 model，都没有则不传 --model（容器内 host.py
+        # 用配置文件里的 model）。fallback 时 worker 换任务的 model。
+        self.model = model
         self.runner = runner if runner is not None else _docker_default_runner
         self.image = image
         self.cpus = str(cpus)
@@ -381,7 +398,7 @@ class DockerExecutor(Executor):
         cmd += list(inner_cmd)
         return cmd
 
-    def _inner_host_command(self, *args):
+    def _inner_host_command(self, *args, model=None):
         """容器内 host.py 命令：python3 /app/host.py [args...]。
 
         config_path 若已给，容器内用 /app/<basename> 指代它——
@@ -389,11 +406,16 @@ class DockerExecutor(Executor):
         本就放在项目目录）；不在代码目录时不静默猜路径，
         由 _inner_host_command 的调用方在传 config 前自行确认。
         这里只做最简单的同名映射，不展开宿主绝对路径进容器。
+        --model 是 host.py 的顶层全局参数，必须放在子命令之前，
+        所以 model 在这里插到 --config 之后、子命令之前。
         """
         cmd = ["python3", "/app/host.py"]
         if self.config_path is not None:
             cmd += ["--config",
                     "/app/" + os.path.basename(self.config_path)]
+        effective = model or self.model
+        if effective:
+            cmd += ["--model", effective]
         cmd += list(args)
         return cmd
 
@@ -515,7 +537,7 @@ class DockerExecutor(Executor):
             raise ValueError("任务缺少 prompt")
         inner = self._inner_host_command(
             "dispatch", "--name", self._host_name(task),
-            "--task", task["prompt"])
+            "--task", task["prompt"], model=task.get("model"))
         repo = task.get("repo")
         if repo:
             inner += ["--repo", repo]
@@ -529,7 +551,7 @@ class DockerExecutor(Executor):
             raise ValueError("任务缺少 prompt")
         inner = self._inner_host_command(
             "run", "--task", self._host_name(task),
-            "--prompt", task["prompt"])
+            "--prompt", task["prompt"], model=task.get("model"))
         return self._run_in_container(task, inner)
 
     def execute(self, task) -> dict:
@@ -660,6 +682,11 @@ class CloudExecutor(Executor):
             "prompt": task.get("prompt") or "",
             "repo_url": task.get("repo"),
         }
+        # 任务带 model 时（worker 每次都设，fallback 时是新模型）
+        # 才发 model 字段：runner 若认就按它跑，不认则忽略；无模型
+        # 的任务请求体与以前完全一致，不含任何密钥。
+        if task.get("model"):
+            payload["model"] = task["model"]
         result = self._post(payload)
         if isinstance(result, dict):
             status = result.get("status")

cloud/tests/test_fallback_retry.py

@@ -0,0 +1,266 @@
+"""真实 fallback 换模型重试单测。
+
+只用注入的假 runner（假 CLI），不跑真实模型回合、不联网、
+不登录任何服务；env 映射也是注入的普通 dict，里面的值只是
+占位用的假名，绝不是真实密钥。
+
+假 runner 的约定：host.py 风格的 JSON 成功时 returncode=0，
+失败且可 fallback 时同样 returncode=0 但 JSON 的 status 字段
+是 "404"/"500"/"timeout"（_result_from_runner 据此把结果标
+failed 并把 host_status 置成该字段，这正是 worker 判定
+fallback 的依据）；returncode 非 0 时 host_status 为空。
+"""
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.cost import CostStore
+    from cloud.models import ModelRouter
+    from cloud.server import init_db
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.cost import CostStore
+    from cloud.models import ModelRouter
+    from cloud.server import init_db
+    from cloud.worker import Worker
+
+
+def _insert_task(db_path, task_id, prompt="do work", status="queued",
+                 project_id="p1"):
+    now = time.time()
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " idempotency_key, depends_on, priority, created_by,"
+            " created_at, updated_at, last_heartbeat)"
+            " VALUES(?, ?, ?, ?, NULL, '[]', 0, 'tester', ?, ?, NULL)",
+            (task_id, project_id, prompt, status, now, now),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _status(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT status FROM tasks WHERE id = ?",
+            (task_id,)).fetchone()
+    finally:
+        conn.close()
+    return row[0] if row else None
+
+
+def _execution(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT error_text, actual_model FROM tasks WHERE id = ?",
+            (task_id,)).fetchone()
+        cost = conn.execute(
+            "SELECT model, fallback_count FROM cost_records"
+            " WHERE task_id = ?",
+            (task_id,)).fetchone()
+        # 测试库里只有一个项目/任务，审计按项目全取：
+        # worker_fallback 的 detail 是“模型 -> 模型”，不含任务 id，
+        # 按 detail LIKE 任务 id 查会漏掉它。
+        audits = [r[0] for r in conn.execute(
+            "SELECT action FROM audit_log WHERE project_id = ?",
+            ("p1",)).fetchall()]
+    finally:
+        conn.close()
+    return row, cost, audits
+
+
+def _host_json(name, status, text):
+    return SimpleNamespace(
+        returncode=0,
+        stdout=json.dumps({"task": name, "status": status, "text": text}),
+        stderr="")
+
+
+def _model_of(cmd):
+    """假 CLI 命令里的 --model 值（没有时返回 None，即配置模型）。"""
+    if "--model" in cmd:
+        return cmd[cmd.index("--model") + 1]
+    return None
+
+
+class FallbackRetryTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.env = {"XCM_API_KEY": "test-placeholder"}
+        self.calls = []
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _worker(self, runner=None, **kwargs):
+        kwargs.setdefault("env", self.env)
+        kwargs.setdefault("retry_backoff_seconds", 0)
+        return Worker(self.db_path, "/fake/host.py",
+                      runner=runner, **kwargs)
+
+    def _recording(self, inner):
+        def fake(cmd):
+            self.calls.append(list(cmd))
+            return inner(cmd)
+        return fake
+
+    def _models_seen(self):
+        """每次调用里实际传给 host.py 的模型（None 表示配置模型）。"""
+        seen = []
+        for cmd in self.calls:
+            if "dispatch" in cmd or "run" in cmd:
+                seen.append(_model_of(cmd))
+        return seen
+
+    def test_primary_404_fallback_succeeds(self):
+        # 主模型 404，配了 fallback 列表：真的换模型跑第二次且成功。
+        def fake(cmd):
+            if _model_of(cmd) == "gpt-6-sol":
+                return _host_json("A", "completed", "done by fallback")
+            return _host_json("A", "404", "primary 404 boom")
+
+        _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(self._models_seen(),
+                         ["gpt-6.1-sol", "gpt-6-sol"])
+        self.assertEqual(len(result["events"]), 1)
+        event = result["events"][0]
+        self.assertEqual(event["type"], "model_fallback")
+        self.assertEqual(event["from_model"], "gpt-6.1-sol")
+        self.assertEqual(event["to_model"], "gpt-6-sol")
+        self.assertEqual(_status(self.db_path, "A"), "completed")
+        row, cost, audits = _execution(self.db_path, "A")
+        self.assertEqual(row[1], "gpt-6-sol")
+        self.assertEqual(cost[0], "gpt-6-sol")
+        self.assertEqual(cost[1], 1)
+        self.assertIn("worker_fallback", audits)
+
+    def test_primary_401_never_fallbacks(self):
+        # 401 绝不 fallback：只调一次，直接 failed，无 fallback 事件。
+        def fake(cmd):
+            return _host_json("A", "401", "bad credentials")
+
+        _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"], [])
+        row, cost, audits = _execution(self.db_path, "A")
+        self.assertEqual(cost[1], 0)
+        self.assertNotIn("worker_fallback", audits)
+
+    def test_no_fallback_list_keeps_old_event_only_behavior(self):
+        # 未配 fallback 列表：老行为——只记事件、直接 failed、调一次。
+        def fake(cmd):
+            return _host_json("A", "404", "primary 404 boom")
+
+        _insert_task(self.db_path, "A")
+        router = ModelRouter(["gpt-6.1-sol", "gpt-6-sol"])
+        worker = self._worker(
+            runner=self._recording(fake),
+            model="gpt-6.1-sol",
+            router=router,
+            max_retries=0)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual(len(self.calls), 1)
+        self.assertEqual(len(result["events"]), 1)
+        self.assertEqual(result["events"][0]["type"], "model_fallback")
+        self.assertEqual(result["actual_model"], "gpt-6.1-sol")
+        row, cost, audits = _execution(self.db_path, "A")
+        self.assertEqual(cost[1], 1)
+        self.assertIn("worker_fallback", audits)
+
+    def test_both_attempts_fail_error_contains_both_reasons(self):
+        # 第二次也失败：error_text 同时含两次原因，fallback_count=1。
+        def fake(cmd):
+            if _model_of(cmd) == "gpt-6-sol":
+                return _host_json("A", "500", "fallback 500 boom")
+            return _host_json("A", "404", "primary 404 boom")
+
+        _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.assertIn("primary 404 boom", result["error"])
+        self.assertIn("fallback 500 boom", result["error"])
+        self.assertIn("gpt-6.1-sol", result["error"])
+        self.assertIn("gpt-6-sol", result["error"])
+        self.assertEqual(len(self.calls), 2)
+        row, cost, audits = _execution(self.db_path, "A")
+        self.assertEqual(cost[1], 1)
+        self.assertEqual(row[1], "gpt-6-sol")
+
+    def test_fallback_at_most_once_even_with_two_models(self):
+        # 配两个 fallback 模型也只试第一个：总共调两次，绝不试第三个。
+        def fake(cmd):
+            return _host_json("A", "429", "always rate limited")
+
+        _insert_task(self.db_path, "A")
+        worker = self._worker(
+            runner=self._recording(fake),
+            model="gpt-6.1-sol",
+            fallback_models=["gpt-6-sol", "gpt-5.6-sol"],
+            max_retries=0)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual(len(self.calls), 2)
+        self.assertEqual(self._models_seen(),
+                         ["gpt-6.1-sol", "gpt-6-sol"])
+        self.assertNotIn("gpt-5.6-sol", self._models_seen())
+
+    def test_timeout_first_attempt_fallback_succeeds(self):
+        # 超时也算可 fallback：第一次 timeout，换模型后成功。
+        def fake(cmd):
+            if _model_of(cmd) == "gpt-6-sol":
+                return _host_json("A", "completed", "done after timeout")
+            return _host_json("A", "timeout", "")
+
+        _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"], "timeout")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -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,