diff --git a/cloud/README.md b/cloud/README.md index 629b3ef..6752683 100644 --- a/cloud/README.md +++ b/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，单个失败只跳过， diff --git a/cloud/sandbox.py b/cloud/sandbox.py index efc3d97..6626fd4 100644 --- a/cloud/sandbox.py +++ b/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.py [--config CFG] [--model M] dispatch --name --task [--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.py [--config CFG] [--model M] run --task --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/ 指代它—— @@ -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") diff --git a/cloud/tests/test_fallback_retry.py b/cloud/tests/test_fallback_retry.py new file mode 100644 index 0000000..0bdde5e --- /dev/null +++ b/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() diff --git a/cloud/worker.py b/cloud/worker.py index e1bac16..7a242e6 100644 --- a/cloud/worker.py +++ b/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,